Function main

testing/postgres-client-tests/python/sqlalchemy-test.py:30–64

()

def main():
    user = sys.argv[1]
    port = int(sys.argv[2])

    conn_string_base = "postgresql+psycopg2://"

    engine = create_engine(conn_string_base +
                           "{user}:password@127.0.0.1:{port}/postgres".format(user=user,
                                                                 port=port)
                           )

    with engine.connect() as con:
        for query_response in QUERY_RESPONSE:
            query = list(query_response.keys())[0]
            exp_results = query_response[query]

            result_proxy = con.execute(text(query))

            try:
                results = result_proxy.fetchall()
                if (results != exp_results) and ("dolt_commit" not in query) and ("dolt_merge" not in query):
                    print("Query:")
                    print(query)
                    print("Expected:")
                    print(exp_results)
                    print("Received:")
                    print(results)
                    sys.exit(1)
            # You can't call fetchall on an insert
            # so we'll just ignore the exception
            except sqlalchemy.exc.ResourceClosedError:
                pass

    con.close()
    sys.exit(0)


main()