"""
Some tests for the database interface and the surrounding helper code.
"""

#c Copyright 2008-2024, the GAVO project <gavo@ari.uni-heidelberg.de>
#c
#c This program is free software, covered by the GNU GPL.  See the
#c COPYING file in the source distribution.


import contextlib
import math
import os
import sys
import time

import numpy

from gavo.helpers import testhelpers

from gavo import base
from gavo import rsc
from gavo import rscdesc
from gavo import svcs
from gavo import utils
from gavo.base import config
from gavo.base import sqlsupport
from gavo.utils import pyfits

import tresc


class GetSQLTest(testhelpers.VerboseTest):
# this is for "dumb" fields.  See also viziertest and pqltest for advanced
# SQL generation from stuff coming in from the web.
	def _assertSQLGenerated(self, ikArgs, inputPars, fragment, sqlPars):
		foundPars = {}
		inputKey = base.makeStruct(svcs.InputKey, **ikArgs)
		foundFragment = base.getSQLForField(inputKey, inputPars, foundPars)
		self.assertEqual(fragment, foundFragment)
		self.assertEqual(sqlPars, foundPars)

	def testSimpleString(self):
		self._assertSQLGenerated(
			{"name": "foo", "type": "text"},
			{"foo": ["';"]},
			'foo=%(foo0)s',	{'foo0': "';"})
	
	def testEmptyString(self):
		self._assertSQLGenerated(
			{"name": "foo", "type": "text"},
			{"foo": [""]},
			'foo=%(foo0)s', {'foo0': ""})

	def testMissingString(self):
		self._assertSQLGenerated(
			{"name": "foo", "type": "text"},
			{},
			None, {})

	def testIntSeq(self):
		self._assertSQLGenerated(
			{"name": "foo", "type": "integer"},
			{"foo": list(map(int, "1 2 3 4".split()))},
			'foo IN %(foo0)s', {'foo0': set([1,2,3,4])})


class ProfileTest(testhelpers.VerboseTest):
	parser = config.ProfileParser("data")

	def testEmptyProfile(self):
		nullProfile = self.parser.parse("test1", None, "")
		self.assertRaisesWithMsg(base.StructureError,
			"Insufficient information to connect to the database in profile 'test1'.",
			base.getDBConnection,
			(nullProfile,))

	def testInvalidProfile(self):
		self.assertRaisesWithMsg(config.ProfileParseError,
			"\"internal\", line 3: unknown setting 'hsot'",
			self.parser.parse,
			("test2", "internal", "database=gavo\nhsot=bar\n"))

	def testWhitespace(self):
		profile = self.parser.parse("foo", None, stream="""
			user = honk

			password="secre t  "
			
			""")
		self.assertEqual(profile.user, "honk")
		self.assertEqual(profile.password, "secre t  ")

	def testUnknownProfile(self):
		self.assertRaisesWithMsg(
			config.ProfileParseError,
			"Requested db profile 'nonexisting' does not exist",
			base.getDBProfile,
			("nonexisting",))


class ConnectionsTest(testhelpers.VerboseTest):
	resources = [("conn", tresc.dbConnection)]

	def testServerVersion(self):
		version = base.getPgVersion()
		self.assertTrue(isinstance(version, tuple))
		self.assertTrue(8<=version[0]<=40)  # let's see if DaCHS lives that long
		self.assertTrue(0<version[1]<30)

	def testConfigReset(self):
		self.conn.configure([("statement_timeout", 0)])
		with self.conn.parameters([("statement_timeout", 20*1000)]):
			self.assertEqual(
				self.conn.getParameter("statement_timeout"), "20s")
		self.assertEqual(self.conn.getParameter("statement_timeout"), '0')

	def testConfigResetBadQuery(self):
		self.conn.configure([("cursor_tuple_fraction", 1)])
		self.conn.commit()

		try:
			with self.conn.parameters([("cursor_tuple_fraction", 0.2)]):
				self.assertEqual(
					self.conn.getParameter("cursor_tuple_fraction"), "0.2")
				self.conn.execute("ghibberish")
		except AssertionError:
			raise
		except:
			self.conn.rollback()

		self.assertEqual(
			self.conn.getParameter("cursor_tuple_fraction"), '1')

	def testConfigResetPlainException(self):
		prevVal = self.conn.getParameter("enable_seqscan")
		newVal = {"on": "off", "off": "on"}[prevVal]
		try:
			with self.conn.parameters([("enable_seqscan", newVal)]):
				self.assertEqual(
					self.conn.getParameter("enable_seqscan"), newVal)
				raise ValueError("test failure.")
		except AssertionError:
			raise
		except ValueError: # raised by myself, I guess
			pass
		self.assertEqual(
			self.conn.getParameter("enable_seqscan"), prevVal)

	def testConfigAutocommitted(self):
		with base.getTableConn() as conn:
			cursor = conn.cursor()
			cursor.execute("SELECT current_setting('statement_timeout')")
			prevVal = list(cursor)[0][0]
			with conn.parameters([("statement_timeout", 34)]):
				cursor.execute("SELECT current_setting('statement_timeout')")
				self.assertEqual(list(cursor)[0][0], "34ms")
			cursor.execute("SELECT current_setting('statement_timeout')")
			self.assertEqual(list(cursor)[0][0], prevVal)
			cursor.close()

	def testNoConnection(self):
		self.assertRaisesWithMsg(
			utils.ReportableError,
			utils.EqualingRE('(?s)Cannot connect to the database server. The database library reported:\n\n.*port 3322.*'),
			base.getDBConnection,
			(config.DBProfile(name="failing", host="localhost", port=3322,
			database="notexiting"),))


@contextlib.contextmanager
def digestedTable(connection, name, columns, values):
	"""a context manager to have a temporary table with ddl and values,
	also providing the whole table after digestion by the database.

	Values is a list of dicts sutable for for INSERT statements.

	ddl is a string that is pasted behind create temp table.

	None of the inputs is validated in any way, so this function must
	never receive untrusted input.
	"""
	connection.execute("CREATE TEMP TABLE %s %s"%(name, columns))
	for row in values:
		connection.execute("INSERT INTO %s (%s) VALUES (%s)"%(
			name, ", ".join(row), ", ".join("%%(%s)s"%s for s in row)),
			row)
	yield list(connection.queryToDicts("SELECT * FROM %s"%name))
	connection.rollback()


class TestTypes(testhelpers.VerboseTest):
	"""Tests for some special adapters we provide.
	"""
	resources = [("conn", tresc.dbConnection)]

	def testNumpyFloat32(self):
		with digestedTable(self.conn, "n_a_t", "(b REAL)",
				[{"b": numpy.float32(0.25)}]) as rows:
			self.assertEqual(rows[0]["b"], 0.25)

	def testNumpyFloat64(self):
		with digestedTable(self.conn, "n_a_t", "(b REAL)",
				[{"b": numpy.float64(0.25)}]) as rows:
			self.assertEqual(rows[0]["b"], 0.25)

	def testNumpyInt8(self):
		with digestedTable(self.conn, "n_a_t", "(b BIGINT)",
				[{"b": numpy.int8(10)}]) as rows:
			self.assertEqual(rows[0]["b"], 10)

	def testNumpyInt64(self):
		with digestedTable(self.conn, "n_a_t", "(b BIGINT)",
				[{"b": numpy.int64(388290102316)}]) as rows:
			self.assertEqual(rows[0]["b"], 388290102316)

	def testNumpyArray(self):
		with digestedTable(self.conn, "n_a_t", "(b integer[])",
				[{"b": [1, 5, 2]}]) as rows:
			self.assertEqual(rows[0]["b"], [1, 5, 2])

	def testIntArray(self):
		with digestedTable(self.conn, "n_a_t", "(b double precision[])",
				[{"b": numpy.array([1.0, 1.5, 2])}]) as rows:
			self.assertEqual(rows[0]["b"], [1.0, 1.5, 2.0])

	def testPyfitsUndefined(self):
		with digestedTable(self.conn, "n_a_t", "(a float, b text)",
				[{"a": pyfits.Undefined(), "b": pyfits.Undefined()}]) as rows:
			self.assertEqual(rows[0]["a"], None)
			self.assertEqual(rows[0]["b"], None)

	def testInt4Range(self):
		with digestedTable(self.conn, "n_a_t", "(a int4range)",
				[{"a": base.NumericRange(2, 3)}]) as rows:
			self.assertEqual(rows[0]["a"], base.NumericRange(2, 3))
	
	def testNaN(self):
		with digestedTable(self.conn, "n_a_t", "(a real)",
				[{"a": float("NaN")}]) as rows:
			self.assertTrue(math.isnan(rows[0]["a"]))

	def testTextArray(self):
		with digestedTable(self.conn, "n_a_t", "(a text[])",
				[{"a": ["ein", "walde"]}]) as rows:
			self.assertEqual(rows[0]["a"], ["ein", "walde"])

	def testVarcharsArray(self):
		with digestedTable(self.conn, "n_a_t", "(a varchar[])",
				[{"a": ["ein", "walde"]}]) as rows:
			self.assertEqual(rows[0]["a"], ["ein", "walde"])


class TestWithTableCreation(testhelpers.VerboseTest):
	resources = [("conn", tresc.dbConnection)]

	tableName = None
	rdId = "test.rd"
	rows = []

	def _assertPrivileges(self, foundPrivs, expPrivs):
		# profile user might not be mentioned in table acl, so retrofit it
		profileUser = base.getDBProfile("admin").user
		expPrivs[profileUser] = foundPrivs[profileUser]
		self.assertEqual(set(foundPrivs), set(expPrivs))
		for role in foundPrivs:
			self.assertEqual(foundPrivs[role], expPrivs[role],
				"Privileges for %s don't match: found %s, expected %s"%(role,
					foundPrivs[role], expPrivs[role]))

	def setUp(self):
		testhelpers.VerboseTest.setUp(self)
		if self.tableName is None:
			return
		self.querier = base.UnmanagedQuerier(connection=self.conn)
		self.tableDef = testhelpers.getTestTable(self.tableName, self.rdId)
		self.table = rsc.TableForDef(self.tableDef, rows=self.rows,
			connection=self.conn, create=True)
		self.conn.commit()

	def tearDown(self):
		if self.tableName is None:
			return
		self.table.drop()
		self.conn.commit()


class TestPrivs(TestWithTableCreation):
	"""Tests for privilege management.
	"""
	tableName = "valSpec"

	resources = [("conn", tresc.dbConnection)]

	def testDefaultPrivileges(self):
		self._assertPrivileges(self.querier.getTablePrivileges(
				self.tableDef.rd.schema, self.tableDef.id),
			self.querier.getACLFromRes(self.tableDef))

	def testBadTableName(self):
		self.assertRaisesWithMsg(ValueError,
			"'botched/table()name' is not a SQL regular identifier",
			base.UnmanagedQuerier(self.conn).hasIndex,
			("botched/table()name", "quatschindex"))


class TestADQLPrivs(TestPrivs):
	"""Tests for privilege management for ADQL-enabled tables.
	"""
	tableName = "adqltable"


class TestRoleSetting(TestPrivs):
	tableName = "privtable"
	rdId = "privtest.rd"

	def setUp(self):
		# We need a private querier here since we must squeeze those
		# users in before TestPriv's setup
		try:
			with base.getWritableAdminConn() as conn:
				conn.execute("create user privtestuser")
				conn.execute("create user testadmin")
		except base.DBError: # probably left over from a previous crash
			sys.stderr.write("Test roles already present."
				" Hope we'll clear them later.\n")
		
		self.profDir = base.getConfig("configDir")
		with open(os.path.join(self.profDir, "privtest"), "w") as f:
			f.write("include dsn\nuser=privtestuser\n")
		with open(os.path.join(self.profDir, "testadmin"), "w") as f:
			f.write("include dsn\nuser=testadmin\n")
	
		TestPrivs.setUp(self)
	
	def tearDown(self):
		TestPrivs.tearDown(self)
		self.conn.execute("drop schema test cascade")
		self.conn.execute("drop user privtestuser")
		self.conn.execute("drop user testadmin")
		self.conn.commit()
		os.unlink(os.path.join(self.profDir, "privtest"))
		os.unlink(os.path.join(self.profDir, "testadmin"))


class AdhocQuerierTest(testhelpers.VerboseTest):
	resources = [("table", tresc.csTestTable), ("conn", tresc.dbConnection)]

	def testBasic(self):
		self.assertEqual(1, len(list(self.conn.query(
			"select * from %s limit 1"%self.table.tableDef.getQName()))))

	def testAdminQuerier(self):
		with base.getWritableAdminConn() as conn:
			self.assertRuns(conn.execute,
				("create role dont",))
			self.assertRuns(conn.execute,
				("drop role dont",))
	
	def testNoAdminQuerier(self):
		with base.getTableConn() as conn:
			self.assertRaises(
				(sqlsupport.ProgrammingError, sqlsupport.InternalError),
				conn.execute,
				"create role dont")

	def testNoInjectionOnGetColumns(self):
		q = base.UnmanagedQuerier(self.conn)
		self.assertRaisesWithMsg(ValueError,
			"'foo; drop tables' is not a SQL regular identifier",
			q.getColumnsFromDB,
			("foo; drop tables",))

	def testGetColumns(self):
		with digestedTable(self.conn, "odd_table",
				"(t text, s smallint, i integer, b bigint,"
				" r real, d double precision, ar real[], c char,"
				" ad double precision[], odd char(20), frotsch bytea[30],"
				" fro bytea, ts timestamp, dt date, ti time,"
				" boo boolean)", []):
			q = base.UnmanagedQuerier(self.conn)
			self.assertEqual(q.getColumnsFromDB("odd_table"), [
				('t', 'text'), ('s', 'int2'), ('i', 'int4'),
				('b', 'int8'), ('r', 'float4'), ('d', 'float8'),
				('ar', '_float4'), ('c', 'bpchar'), ('ad', '_float8'),
				('odd', 'bpchar'), ('frotsch', '_bytea'), ('fro', 'bytea'),
				('ts', 'timestamp'), ('dt', 'date'), ('ti', 'time'),
				('boo', 'bool')])

	def testRowEstimateGood(self):
		estimate = base.UnmanagedQuerier(self.conn
			).getRowEstimate("tap_schema.tables")
		self.assertTrue(isinstance(estimate, int))
		self.assertTrue(0<=estimate<=200)

	def testRowEstimateBad(self):
		self.assertRaisesWithMsg(base.DBError,
			utils.EqualingRE('relation "tap_schema.goneaway" does not exist\n.*'),
			base.UnmanagedQuerier(self.conn).getRowEstimate,
			("tap_schema.goneaway",))

	def testOIDGood(self):
		oid = base.UnmanagedQuerier(self.conn).getOIDForTable("tap_schema.tables")
		self.assertTrue(isinstance(oid, int))
		self.assertTrue(1000<oid)

	def testOIDBad(self):
		self.assertRaisesWithMsg(base.DBError,
			utils.EqualingRE('schema "xanadu" does not exist.*'),
			base.UnmanagedQuerier(self.conn).getOIDForTable,
			("xanadu.goneaway",))

	def testPrivsForNonExisting(self):
		self.assertEqual(
			base.UnmanagedQuerier(self.conn
				).getTablePrivileges("xanadu", "goneaway"),
			{})


class DBOpsTest(testhelpers.VerboseTest):
	resources = [("conn", tresc.dbConnection)]

	def testDroppingMissing(self):
		self.assertEqual(
			base.UnmanagedQuerier(self.conn).dropTable("xanadu.goneaway"),
			None)


class TestMetaTable(TestWithTableCreation):
	tableName = "typesTable"

	def testDcTablesEntry(self):
		with base.getTableConn() as conn:
			res = list(conn.query("select * from dc.tablemeta where tableName=%(n)s",
				{"n": self.tableDef.getQName()}))
		qName, srcRd, td, rd, adql, nrows = res[0]
		self.assertEqual(qName, 'test.typesTable')
		self.assertEqual(srcRd.split("/")[-1], 'test')
		self.assertEqual(adql, True)


class TestMetaTableADQL(TestWithTableCreation):
	tableName = "adqltable"

	def testDcTablesEntry(self):
		res = list(self.conn.query(
			"select * from dc.tablemeta where tableName=%(n)s",
			{"n": self.tableDef.getQName()}))
		qName, srcRd, td, rd, adql, nrows = res[0]
		self.assertEqual(qName, 'test.adqltable')
		self.assertEqual(srcRd.split("/")[-1], 'test')
		self.assertEqual(adql, True)


class TestWithDataImport(testhelpers.VerboseTest):
	"""base class for tests importing data up front.

	You need to set the ddId, which must point into test.rd
	"""
	resources = [("connection", tresc.dbConnection)]

	def setUp(self):
		testhelpers.VerboseTest.setUp(self)
		dd = testhelpers.getTestRD().getById(self.ddId)
		self.data = rsc.makeData(dd, connection=self.connection)
	

class TestPreIndexSQLRunning(TestWithDataImport):
	"""tests for dbtables running preIndexSQL scripts.
	"""
	ddId = "import_sqlscript"

	def testScriptRan(self):
		ct = list(self.connection.query(
			"select count(*) from test.sqlscript"))[0][0]
		self.assertEqual(ct, 3)


class TestPreIndexPythonRunning(TestWithDataImport):
	"""tests for dbtables running preIndexSQL scripts.
	"""
	ddId = "import_pythonscript"

	def testScriptRan(self):
		ct = list(self.connection.query("select * from test.pythonscript"))[0][0]
		self.assertEqual(ct, 123)


class TestQueryExpands(TestWithTableCreation):
	"""tests for expansion of macros in dbtable's query.
	"""
	tableName = "adqltable"

	def testExpandedQuery(self):
		self.table.connection.execute(
			self.table.expand(
				r"insert into \qName (\colNames) values (-133.0)"))
		self.assertEqual(
			len(list(self.table.iterQuery([self.tableDef.getColumnByName("foo")],
				"foo=-133.0"))),
			1)

	def testBadMacroRaises(self):
		self.assertRaises(base.MacroError, self.table.expand, r"\monkmacrobad")


class NewlyCreatedTest(testhelpers.VerboseTest):
	resources = [("connection", tresc.dbConnection)]

	def testNewlyCreated(self):
		rd = base.parseFromString(rscdesc.RD,
			'<resource schema="test"><table id="newly_created" onDisk="true"/>'
			'</resource>')
		rd.sourceId  = "nct"
		td = rd.getById("newly_created")
		rsc.TableForDef(td, create=False, connection=self.connection).drop()
		self.connection.commit()

		try:
			t = rsc.TableForDef(td, create=True, connection=self.connection)
			self.assertEqual(t.newlyCreated, True)
			t = rsc.TableForDef(td, create=True, connection=self.connection)
			self.assertEqual(t.newlyCreated, False)
		finally:
			self.connection.rollback()


class StatisticsTargetTest(testhelpers.VerboseTest):
	resources = [("connection", tresc.dbConnection)]

	def testSetStatTarget(self):
		rd = base.parseFromString(rscdesc.RD,
			"""<resource schema="test"><table id="statatest" onDisk="True">
			<column name="quoted/x X"><property name="statisticsTarget">4</property>
			</column></table>
			<data id="import"><sources item="500"/><embeddedGrammar><iterator>
			<code>
				for i in range(int(self.sourceToken)):yield {"x": i}
			</code></iterator></embeddedGrammar>
			<make table="statatest">
				<rowmaker><apply><code>result["x X"]=@x</code></apply></rowmaker>
			</make></data></resource>""")
		rd.sourceId = "internal"
		_ = rsc.makeData(rd.getById("import"),
			connection=self.connection)

		stats = list(
			self.connection.query("SELECT histogram_bounds from pg_stats"
			" WHERE tablename='statatest' and attname='x X'"))
		# The array isn't parsed right now, so I'm counting commas in the
		# literal. If it *is* parsed and you get an array here: Great!
		self.assertTrue(len(stats[0][0].split(","))<10,
			"Postgres made many more histogram bins than our statistics target")


class TestIgnoreFrom(testhelpers.VerboseTest):
	resources = [("connection", tresc.dbConnection)]

	def testWorksWithoutTable(self):
		rsc.TableForDef(testhelpers.getTestRD().getById("prodtest"),
			connection=self.connection, create=False).drop()
		self.data = rsc.makeData(
			testhelpers.getTestRD().getById("productimport"),
			rsc.getParseOptions(keepGoing=True),
			connection=self.connection)
		self.assertEqual(
			set(self.connection.query("select object from test.prodtest")),
			set([('gabriel',), ('michael',)]))

	def testWorksWithTable(self):
		dd = testhelpers.getTestRD().getById("productimport")

		data1 = rsc.makeData(dd,
			rsc.getParseOptions(keepGoing=True),
			connection=self.connection)
		self.assertEqual(data1.nAffected, 2)

		modDD = dd.change(updating=True)

		data2 = rsc.makeData(modDD,
			rsc.getParseOptions(keepGoing=True),
			connection=self.connection)
		self.assertEqual(data2.nAffected, 0)


class InformationSchemaTest(testhelpers.VerboseTest):
	resources = [("connection", tresc.dbConnection)]

	def testNoTableType(self):
		q = sqlsupport.UnmanagedQuerier(self.connection)
		self.assertEqual(q.getTableType("thereareno.monsters"),
			None)

	def _assertProperlyManaged(self,
			creationStatement,
			tableName,
			tableType):
		q = sqlsupport.UnmanagedQuerier(self.connection)
		# if things go seriously wrong in one of these tests, you may
		# have to manually remove the relations created here.
		self.connection.execute(creationStatement)
		try:
			self.assertEqual(q.getTableType(tableName), tableType)
		finally:
			q.dropTable(tableName)
		self.assertEqual(q.getTableType(tableName), None)

	def testTempTableManagement(self):
		self._assertProperlyManaged(
			"CREATE TEMP TABLE klutz (an INTEGER)",
			"klutz",
			"LOCAL TEMPORARY")

	def testNormalTableManagement(self):
		self._assertProperlyManaged(
			"CREATE TABLE tap_schema.klutz (an INTEGER)",
			"tap_schema.klutz",
			"BASE TABLE")

	def testPublicTableManagement(self):
		self._assertProperlyManaged(
			"CREATE TABLE klutz (an INTEGER)",
			"klutz",
			"BASE TABLE")

	def testViewManagement(self):
		self._assertProperlyManaged(
			"CREATE VIEW klutz AS (SELECT 1 as an)",
			"klutz",
			"VIEW")

	def testMaterializedViewManagement(self):
		self._assertProperlyManaged(
			"CREATE MATERIALIZED VIEW IF NOT EXISTS klutz AS (SELECT 1 as an)",
			"klutz",
			"MATERIALIZED VIEW")

	def testSchemaExists(self):
		querier = base.UnmanagedQuerier(self.connection)
		self.assertTrue(querier.schemaExists("TaP_ScHeMa"))


class DerivedViewsDroppingTest(testhelpers.VerboseTest):
	resources = [("conn", tresc.dbConnection)]

	def testViewScriptRunning(self):
		with testhelpers.testFile("viewdropping.rd",
				inDir=base.getConfig("inputsDir"),
				content=r"""<resource schema="test">
					<table id="to_derive" onDisk="True">
						<column name="dist"/>
					</table>
					<table id="is_derived" onDisk="True">
						<column original="to_derive.dist"/>
						<viewStatement>
							CREATE VIEW \qName AS (SELECT \colNames from \schema.to_derive)
						</viewStatement>
						<script type="beforeDrop" lang="python">
							table.tableDef.rd.properlyDropped = True
						</script>
					</table>
					<data id="import">
						<make table="to_derive"/>
						<make table="is_derived"/>
					</data>
				</resource>"""):
			rd = base.caches.getRD("viewdropping")
			data = rsc.makeData(rd.getById("import"), connection=self.conn)
			data.tables["to_derive"].drop()
			self.assertTrue(hasattr(rd, "properlyDropped"),
				"Derived view's todrop script hasn't run")


class NonblockingQueryTest(testhelpers.VerboseTest):
	resources = [("conn", tresc.dbConnection)]

	def testRunning(self):
		q = sqlsupport.NonBlockingQuery(self.conn, "select"
				" table_name from TAP_SCHEMA.tables where table_name like %(nt)s"
				" order by table_name limit 1",
				{"nt": "tap_schema.%"})
		with q:
			for i in range(100):
				if q.result is not None:
					break
				time.sleep(0.01)
			else:
				raise AssertionError("Nonblocking query does not finish")
		
		self.assertEqual(q.result,  [('tap_schema.columns',)])
		self.assertFalse(q.thread.is_alive())

	def testFailing(self):
		q = sqlsupport.NonBlockingQuery(self.conn, "select * from junk")
		try:
			with q:
				while q.result is None:
					time.sleep(0.01)
		except Exception as ex:
			self.assertEqual(
				ex.pgerror[:38],
				'ERROR:  relation "junk" does not exist')
		else:
			raise AssertionError("No exception seen for failing query.")

		self.assertFalse(q.thread.is_alive())

	def testAborting(self):
		q = sqlsupport.NonBlockingQuery(self.conn, "select (select avg(asin(sqrt(x)/224.0)) from generate_series(1, whomp) as x) as q from generate_series(1, 50000) as whomp")
		with q:
			q.abort()
		self.assertFalse(q.thread.is_alive())


class DBMetaTest(testhelpers.VerboseTest):
	def testVersionCurrent(self):
		from gavo.user import upgrade
		self.assertEqual(
			int(base.getDBMeta("schemaversion")),
			upgrade.CURRENT_SCHEMAVERSION)
	
	def testBadGetting(self):
		self.assertRaisesWithMsg(
			KeyError,
			"'xanadu'",
			base.getDBMeta,
			("xanadu",))


class GeometryCastTests(testhelpers.VerboseTest):
# since what's tested here is defined in //adql, this should perhaps go to
# adqltest?  Perhaps we should pull all geometry-related ADQL testing into
# a test suite of its own?
	resources = [("conn", tresc.dbConnection)]

	def testGoodPointCast(self):
		res = list(self.conn.query("select cast_to_point('270 45')"))[0][0]
		lon, lat = res.asDALI()
		self.assertAlmostEqual(lon, 270)
		self.assertAlmostEqual(lat, 45)

	def testWrongPointCard(self):
		self.assertRaisesWithMsg(base.DBError,
			"Not a DALI POINT literal: 270 45 23\nCONTEXT:  PL/pgSQL function cast_to_point(text) line 8 at RAISE\n",
			list,
			(self.conn.query("select cast_to_point('270 45 23')"),))

	def testWrongPointNumberSyntax(self):
		self.assertRaisesWithMsg(base.DBError,
			"invalid input syntax for type double precision: \"abc\"\nCONTEXT:  PL/pgSQL function cast_to_point(text) line 10 at RETURN\n",
			list,
			(self.conn.query("select cast_to_point('abc def')"),))

	def testGoodCircleCast(self):
		res = list(self.conn.query("select cast_to_circle('110.3 -3e-2 3e-6')"
			))[0][0]
		lon, lat, radius = res.asDALI()
		self.assertAlmostEqual(lon, 110.3)
		self.assertAlmostEqual(lat, -3e-2)
		self.assertAlmostEqual(radius, 3e-6)

	def testWrongCircleCard(self):
		self.assertRaisesWithMsg(base.DBError,
			"Not a DALI CIRCLE literal: 270 45\nCONTEXT:  PL/pgSQL function cast_to_circle(text) line 8 at RAISE\n",
			list,
			(self.conn.query("select cast_to_circle('270 45')"),))

	def testGoodPolyCast(self):
		res = list(self.conn.query(
				"select cast_to_polygon('270 45 271 45 270.5 44')"
			))[0][0]
		coos = res.asDALI()
		self.assertAlmostEqual(coos[4], 270.5)
		self.assertAlmostEqual(coos[5], 44)

		coverage = res.asSMoc(7).asASCII()
		# in case the following assertion changes, it's all possible
		# there's a better healpix library to praise...
		self.assertEqual(coverage, "7/60095 60138 7/")

	def testWrongPolyCard(self):
		self.assertRaisesWithMsg(base.DBError,
			"Not a DALI POLYGON literal: 270 45 271 45 270.5 44 28\nCONTEXT:  PL/pgSQL function cast_to_polygon(text) line 14 at RAISE\n",
			list,
			(self.conn.query("select cast_to_polygon('270 45 271 45 270.5 44 28')"),))


if __name__=="__main__":
	testhelpers.main(AdhocQuerierTest)