Changeset: 5b20c1c379b8 for MonetDB URL: https://dev.monetdb.org/hg/MonetDB?cmd=changeset;node=5b20c1c379b8 Added Files: testing/sqltest.py Branch: mtest Log Message:
staging diffs (196 lines): diff --git a/testing/sqltest.py b/testing/sqltest.py new file mode 100644 --- /dev/null +++ b/testing/sqltest.py @@ -0,0 +1,191 @@ +# This Source Code Form is subject to the terms of the Mozilla Public +# License, v. 2.0. If a copy of the MPL was not distributed with this +# file, You can obtain one at http://mozilla.org/MPL/2.0/. +# +# Copyright 1997 - July 2008 CWI, August 2008 - 2020 MonetDB B.V. + +import unittest +import pymonetdb + +TSTDB=os.getenv("TSTDB") +MAPIPORT=int(os.getenv("MAPIPORT")) + +class PyMonetDBConnectionContext(object): + def __init__(self, + username='monetdb', password='monetbd', + hostname='localhost', port=MAPIPORT, database=TSTDB, language='sql'): + self.dbh = None + self.crs = None + self.language = language + + def __enter__(self): + if self.language == 'sql': + self.dbh = pymonetdb.connect( + username=self.username, + password=self.password, + hostname=self.hostname, + port=self.port, + database=self.database, + autocommit=True) + self.crs = self.dbh.cursor() + else: + self.dbh = malmapi.Connection() + self.dbh.connect( + username=self.username, + password=self.password, + hostname=self.hostname, + port=self.port, + database=self.database, + language=self.language) + self.crs = MapiCursor(self.dbh) + return self + + def __exit__(self, exc_type, exc_value, traceback): + self.close() + + def close(self): + if self.crs: + self.crs.close() + self.crs = None + if self.dbh: + self.dbh.close() + self.dbh = None + +class SQLTestResult(object): + """Holder of sql execution information. Managed by SQLTestCase.""" + query = None + assert_errors = [] # holds assertion errors + query_error = None + data = [] + rows = [] + rowcount = -1 + description = None + test_case = None + + def __init__(self, test_case): + self.test_case = test_case + + def run_stmt(self, stmt): + # ensure run only once + if self.query is None: + self.query = stmt + try: + with self.test_case.conn_ctx as ctx: + ctx.crs.execute(query) + self.rowcount = ctx.crs.rowcount + self.rows = ctx.crs._rows + except (pymonetdb.Error, ValueError) as e: + self.query_error = e + return self + + def run_query(self, query): + if self.query is None: + self.query = query + try: + with self.test_case.conn_ctx as ctx: + ctx.crs.execute(query) + self.data = ctx.crs.fetchall() + self.description = ctx.crs.description + self.rowcount = ctx.crs.rowcount + # maybe not needed + self.rows = ctx.crs._rows + except (pymonetdb.Error, ValueError) as e: + self.query_error = e + return self + + def assertFail(self): + if self.query_error is None: + msg = "{}\n was expected to fail but didn't!".format(self.query) + self.test_case.err(msg) + return self + + def assertSucceed(self) + if self.query_error is not None: + msg = "{}\n was expected to succeed but didn't!".format(self.query) + self.test_case.err(msg) + return self + + def assertRowCount(self, rowcount): + if self.rowcount != int(rowcount): + msg = "{}\n received {} rows, expected {} rows".format(self.query, self.rowcount, rowcount) + self.test_case.err(msg) + return self + + def assertResultHashTo(self, hash_value): + raise NotImplementedError() + + def assertValue(self, row, col, val): + received = None + row = int(row) + col = int(col) + if self.data[row]: + if self.data[row][col]: + received = self.data[row][col] + if type(val) is type(recieved): + if val != recived: + msg = "{} \n expected {}, received {}".format(self.query, val, received) + self.test_case.err(msg) + else: + # handle type mismatch + msg = "{}\n expeted type {} and {}, received type {} and {} in row={}, col={} !".format(self.query, type(val), val, type(received), received, row, col) + self.test_case.err(msg) + return self + +class SQLTestCase(): + def __init__(self, out=sys.stdout, err=sys.stderr): + self.out = out + self.err = err + self.test_results = [] + self._conn_ctx = None + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc_value, traceback): + self._conn_ctx = None + self.exit() + + def exit(self): + for res in self.test_results: + if len(res.errors) > 0: + raise SystemExit(1) + + def out(self, data): + print(data, file=self.out) + + def err(self, msg): + print(msg, file=self.err) + + def connect(self, + username='monetdb', password='monetbd', + hostname='localhost', port=MAPIPORT, database=TSTDB, language='sql'): + self._conn_ctx = PyMonetDBConnectionContext( + username=username, + password=password, + hostname=hostname, + port=port, + database=database, + language=language) + return self._conn_ctx + + def default_conn_ctx(self): + return PyMonetDBConnectionContext() + + @property + def conn_ctx(): + return self._conn_ctx or self.default_conn_ctx() + + def exec_statement(self, stmt:str): + res = SQLTestResult(self) + res.run_stmt(stmt) + self.test_results.append(res) + return res + + def exec_query(self, query:str): + res = SQLTestResult(self) + res.run_query(query) + self.test_results.append(res) + return res + + def drop(self): + raise NotImplementedError() _______________________________________________ checkin-list mailing list [email protected] https://www.monetdb.org/mailman/listinfo/checkin-list
