1import os
2
3from ...engine import url as sa_url
4from ...testing.provision import create_db
5from ...testing.provision import drop_db
6from ...testing.provision import follower_url_from_main
7from ...testing.provision import log
8from ...testing.provision import post_configure_engine
9from ...testing.provision import run_reap_dbs
10from ...testing.provision import temp_table_keyword_args
11
12
13@follower_url_from_main.for_db("sqlite")
14def _sqlite_follower_url_from_main(url, ident):
15    url = sa_url.make_url(url)
16    if not url.database or url.database == ":memory:":
17        return url
18    else:
19        return sa_url.make_url("sqlite:///%s.db" % ident)
20
21
22@post_configure_engine.for_db("sqlite")
23def _sqlite_post_configure_engine(url, engine, follower_ident):
24    from sqlalchemy import event
25
26    @event.listens_for(engine, "connect")
27    def connect(dbapi_connection, connection_record):
28        # use file DBs in all cases, memory acts kind of strangely
29        # as an attached
30        if not follower_ident:
31            # note this test_schema.db gets created for all test runs.
32            # there's not any dedicated cleanup step for it.  it in some
33            # ways corresponds to the "test.test_schema" schema that's
34            # expected to be already present, so for now it just stays
35            # in a given checkout directory.
36            dbapi_connection.execute(
37                'ATTACH DATABASE "test_schema.db" AS test_schema'
38            )
39        else:
40            dbapi_connection.execute(
41                'ATTACH DATABASE "%s_test_schema.db" AS test_schema'
42                % follower_ident
43            )
44
45
46@create_db.for_db("sqlite")
47def _sqlite_create_db(cfg, eng, ident):
48    pass
49
50
51@drop_db.for_db("sqlite")
52def _sqlite_drop_db(cfg, eng, ident):
53    for path in ["%s.db" % ident, "%s_test_schema.db" % ident]:
54        if os.path.exists(path):
55            log.info("deleting SQLite database file: %s" % path)
56            os.remove(path)
57
58
59@temp_table_keyword_args.for_db("sqlite")
60def _sqlite_temp_table_keyword_args(cfg, eng):
61    return {"prefixes": ["TEMPORARY"]}
62
63
64@run_reap_dbs.for_db("sqlite")
65def _reap_sqlite_dbs(url, idents):
66    log.info("db reaper connecting to %r", url)
67
68    log.info("identifiers in file: %s", ", ".join(idents))
69    for ident in idents:
70        # we don't have a config so we can't call _sqlite_drop_db due to the
71        # decorator
72        for path in ["%s.db" % ident, "%s_test_schema.db" % ident]:
73            if os.path.exists(path):
74                log.info("deleting SQLite database file: %s" % path)
75                os.remove(path)
76