"""
Tests for rsc.XTable
"""

#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 datetime

from gavo.helpers import testhelpers

from gavo import base
from gavo import rsc
from gavo import rscdef
from gavo import rscdesc
from gavo import svcs
from gavo.protocols import tap
from gavo.user import validation
from gavo.stc import dm

import tresc

class MemoryPrimaryKeyTest(testhelpers.VerboseTest):
	"""tests for handling of primary keys on in-memory tables.
	"""
	def testAtomicPrimary(self):
		td = base.parseFromString(rscdef.TableDef, '<table><column name="x" type='
			'"text"/><column name="y" type="text"/><primary>x</primary></table>')
		t = rsc.TableForDef(td,
			rows=[{"x": "aba", "y": "bab"}, {"x": "xyx", "y": "yxy"}])
		self.assertEqual(t.getRow("aba"), {"x": "aba", "y": "bab"})
		self.assertRaises(KeyError, t.getRow)
		self.assertRaises(KeyError, t.getRow, "wommo")
		t.addRow({"x": "wommo"})
		self.assertEqual(t.getRow("wommo"), {"x": "wommo"})
		t = rsc.TableForDef(td,
			rows=[{"x": "aba", "y": "bab"}, {"x": "xyx", "y": "yxy"}],
			suppressIndex=True)
		self.assertRaises(ValueError, t.getRow, "aba")

	def testCompoundPrimary(self):
		td = base.parseFromString(rscdef.TableDef, '<table><column name="x" type='
			'"text"/><column name="y" type="text"/><primary>x,y</primary></table>')
		t = rsc.TableForDef(td, rows=
			[{"x": "aba", "y": "bab"}, {"x": "xyx", "y": "yxy"}], create=True)
		self.assertEqual(t.getRow("aba", "bab"), {"x": "aba", "y": "bab"})
		self.assertRaises(KeyError, t.getRow, "bab", "aba")
		self.assertRaises(KeyError, t.getRow, "xyx")


class UniqueForcedTest(tresc.TestWithDBConnection):
	"""tests for correct handling of duplicate keys when uniqueness is enforced.
	"""
	_tt = '<resource schema="test"><table id="bla" %s</resource>'

	def _makeTD(self, tdLiteral):
		rd = base.parseFromString(rscdesc.RD,
			self._tt%tdLiteral)
		return rd.getById("bla")

	def testCheckOk(self):
		td = self._makeTD('dupePolicy="check"><column name="x" type='
			'"text"/><column name="y" type="text"/><primary>x</primary></table>')
		t = rsc.TableForDef(td, nometa=True, rows=
			[{"x": "aba", "y": "bab"}, {"x": "xyx", "y": "yxy"}],
			connection=self.conn, create=True)
		self.assertEqual(t.getRow("aba"), {"x": "aba", "y": "bab"})
		t.addRow({"x": "aba", "y": "bab"})
		self.assertEqual(t.getRow("aba"), {"x": "aba", "y": "bab"})
	
	def testCheckRaises(self):
		td = self._makeTD('dupePolicy="check"><column name="x" type='
			'"text"/><column name="y" type="text"/><primary>x</primary></table>')
		self.assertRaises(base.ValidationError, rsc.TableForDef, td, nometa=True,
			rows=[{"x": "aba", "y": "bab"}, {"x": "aba", "y": "yxy"}],
			connection=self.conn, create=True)
		t = rsc.TableForDef(td, rows=
			[{"x": "aba", "y": "bab"}, {"x": "xyx", "y": "yxy"}],
			connection=self.conn, create=True)
		self.assertRaisesWithMsg(base.ValidationError,
			"Field y: Differing rows for primary key ('aba',); bax vs. bab",
			t.addRow, ({"x": "aba", "y": "bax"},))

	def testDrop(self):
		td = self._makeTD('dupePolicy="drop">'
			'<column name="x" type="text"/><column name="y" type="text"/>'
			'<primary>x</primary></table>')
		t = rsc.TableForDef(td, nometa=True, rows=
			[{"x": "aba", "y": "bab"}, {"x": "xyx", "y": "yxy"}],
			connection=self.conn, create=True)
		self.assertEqual(t.getRow("aba"), {"x": "aba", "y": "bab"})
		t.addRow({"x": "aba", "y": "bax"})
		self.assertEqual(t.getRow("aba"), {"x": "aba", "y": "bab"})

	def testDropOld(self):
# in memory, that's actually the same thing as overwrite, but never mind
		td = self._makeTD('>'
			'<dupePolicy>overwrite</dupePolicy><column name="x" type='
			'"text"/><column name="y" type="text"/><primary>x</primary></table>')
		t = rsc.TableForDef(td, nometa=True, rows=
			[{"x": "aba", "y": "bab"}, {"x": "xyx", "y": "yxy"}],
			connection=self.conn, create=True)
		self.assertEqual(t.getRow("aba"), {"x": "aba", "y": "bab"})
		t.addRow({"x": "aba", "y": "bax"})
		self.assertEqual(t.getRow("aba"), {"x": "aba", "y": "bax"})

	def testOverwrite(self):
		td = self._makeTD('>'
			'<dupePolicy>overwrite</dupePolicy><column name="x" type='
			'"text"/><column name="y" type="text"/><primary>x</primary></table>')
		t = rsc.TableForDef(td, nometa=True, rows=
			[{"x": "aba", "y": "bab"}, {"x": "xyx", "y": "yxy"}],
			connection=self.conn, create=True)
		self.assertEqual(t.getRow("aba"), {"x": "aba", "y": "bab"})
		t.addRow({"x": "aba", "y": "bax"})
		self.assertEqual(t.getRow("aba"), {"x": "aba", "y": "bax"})

	def testCompoundKey(self):
		td = self._makeTD('>'
			'<dupePolicy>overwrite</dupePolicy><column name="x" type='
			'"text"/><column name="n" type="integer"/><column name="y" type="text"/>'
			'<primary>x,n</primary></table>')
		t = rsc.TableForDef(td, nometa=True, rows=
			[{"x": "aba", "n": 1, "y": "bab"}, {"x": "aba", "n":2, "y": "yxy"}],
			connection=self.conn, create=True)
		t.addRow({"x": "aba", "n": 3, "y": "bab"})
		t.addRow({"x": "aba", "n": 1, "y": "aba"})
		self.assertEqual(t.getRow("aba", 1), {"x": "aba", "n": 1, "y": "aba"})

	def testWithNull(self):
		td = self._makeTD('dupePolicy="check"><column name="x" type='
			'"text"/><column name="y" type="text"/><primary>x</primary></table>')
		t = rsc.TableForDef(td, nometa=True, rows=
			[{"x": "aba", "y": None}, {"x": "xyx", "y": None}],
			connection=self.conn, create=True)
		self.assertEqual(t.getRow("aba"), {"x": "aba", "y": None})
		t.addRow({"x": "aba", "y": None})
		self.assertEqual(t.getRow("aba"), {"x": "aba", "y": None})
		self.assertRaises(Exception,
			t.addRow,
			({"x": "aba", "y": "bab"}))


class WrapTest(testhelpers.VerboseTest):
	"""tests for wrapping of Tables in Data.
	"""
	def testWithRd(self):
		table = rsc.TableForDef(
			testhelpers.getTestRD().getById("typesTable").change(onDisk=False))
		data = rsc.wrapTable(table)
		self.assertEqual(len(data.tables), 1)
		self.assertEqual(str(data.getMeta("test.inRd")), "from Rd")
	
	def testWithRdSource(self):
		origTable = rsc.TableForDef(
			testhelpers.getTestRD().getById("typesTable").change(onDisk=False))
		newTD = rscdef.makeTDForColumns(
			"copy", origTable.tableDef.columns)
		data = rsc.wrapTable(rsc.TableForDef(newTD), origTable.tableDef)
		self.assertEqual(data.dd.rd.schema, "test")
		self.assertEqual(data.getPrimaryTable().tableDef.rd.schema, "test")


class DBUniqueForcedTest(UniqueForcedTest):
	"""tests for correct handling of duplicate keys when uniqueness is enforced
	in DB tables
	"""
	_tt = '<resource schema="test"><table onDisk="True" id="bla" %s</resource>'

	def testCheckRaises(self):
# For these, semantics between DB tables and InMemoryTables differ --
# DBTables currently raise other Exceptions, and partially at different
# times.
		td = self._makeTD('dupePolicy="check"><column name="x" type='
			'"text"/><column name="y" type="text"/><primary>x</primary></table>')
		t = rsc.TableForDef(td, nometa=True, rows=
			[{"x": "aba", "y": "bab"}, {"x": "xyx", "y": "yxy"}],
			connection=self.conn, create=True)
		self.assertRaisesWithMsg(base.DBError,
			'Overwrite attempt with different tuple with check policy (aba,bab) (aba,bax).\nCONTEXT:  PL/pgSQL function "handle_dupe_test.bla"() line 9 at RAISE\n',
			t.addRow,
			({"x": "aba", "y": "bax"},))

	def testDropOld(self):
# this test is quite complex because the difference between overwrite
# and dropOld is only exibited when there's a foreign key relationship
		rd = base.parseFromString(rscdesc.RD, """<resource schema="test">
			<table onDisk="True" id="a" dupePolicy="dropOld"
					primary="x">
				<column name="x" type="text"/>
			</table>
			<table onDisk="True" id="b">
				<foreignKey source="x" inTable="a"/>
				<column original="a.x"/><column name="y" type="text"/>
			</table>
			<data id="bla">
				<sources items="x"/>
				<embeddedGrammar isDispatching="True"><iterator>
					<code>
						yield ('a', {'x': 'old'})
						yield ('b', {'x': 'old', 'y': 'fromold'})
					</code>
				</iterator></embeddedGrammar>
				<make table="a" role="a"/><make table="b" role="b"/>
			</data></resource>""")
		rd.sourceId = "test/internal"
		try:
			data = rsc.makeData(rd.getById("bla"), connection=self.conn)

			tdb = rd.getById("b")
			tb = rsc.TableForDef(tdb, connection=self.conn, create=True)
			self.assertEqual(list(tb.iterQuery(tdb, "")),
				[{'x': 'old', 'y': 'fromold'}])

			tda = rd.getById("a")
			ta = rsc.TableForDef(tda, connection=self.conn)
			ta.addRow({'x': 'old'})
			self.assertEqual(list(ta.iterQuery(tda, "")), [{'x': 'old'}])

			# and this is the kicker: the row in b has been dropped
			self.assertEqual(list(tb.iterQuery(tdb, "")), [])

			data.drop(data.dd, connection=self.conn)
		except:
			self.conn.rollback()
			raise
		finally:
			self.conn.commit()


class DBTableTest(tresc.TestWithDBConnection):
	"""tests for creation, filling and takedown of DB tables.
	"""
	def _getRD(self):
		return base.parseFromString(rscdesc.RD, '<resource schema="testing">'
			'<table id="xY" onDisk="True">'
			'<column name="x" type="integer"/>'
			'<column name="y" type="text"/><primary>x</primary></table>'
			'</resource>')

	def testCreation(self):
		td = self._getRD().getTableDefById("xY")
		table = rsc.TableForDef(td, connection=self.conn, nometa=True)
		self.assertTrue(table.exists())
		table.drop()
		self.assertTrue(table.getTableType(td.getQName()) is None)

	def testValidationFailure(self):
		td = self._getRD().getTableDefById("xY")
		table = rsc.TableForDef(td, connection=self.conn, nometa=True)
		try:
			self.conn.execute("alter table testing.xy drop column x")
			self.assertRaisesWithMsg(base.DataError,
				'Table testing.xY: mismatching name of x (DB: y); column y:'
				' No matches in DB',
				table.ensureOnDiskMatches,
				())
		finally:
			table.drop()

	def testFilling(self):
		td = self._getRD().getTableDefById("xY")
		table = rsc.TableForDef(td, nometa=True, connection=self.conn)
		table.recreate()
		table.addRow({'x': 100, 'y': "abc"})
		self.assertEqual([{'x': 100, 'y': "abc"}], [row for row in table])
	
	def testFeeding(self):
		td = self._getRD().getTableDefById("xY")
		table = rsc.TableForDef(td, nometa=True, connection=self.conn)
		table.recreate()
		table.feedRows([
			{'x': 100, 'y': "abc"},
			{'x': 200, 'y': "ab"},
			{'x': 50, 'y': "ab"}])
		self.assertEqual(3, len([row for row in table]))

	def testResolutionLiterally(self):
		td = testhelpers.getTestTable("ADQLgeo")
		table = rsc.TableForDef(td, connection=self.conn, create=True)
		td = base.getTableDefForTable(self.conn, "test.ADQLgeo")
		self.assertEqual(td.columns[0].name, "row_id")
		table.drop()
		self.assertRaisesWithMsg(
			base.NotFoundError,
			"table 'test.ADQLgeo' could not be located in published tables",
			base.getTableDefForTable,
			(self.conn, "test.ADQLgeo"))

	def testResolutionNocase(self):
		td = testhelpers.getTestTable("ADQLgeo")
		table = rsc.TableForDef(td, connection=self.conn, create=True)
		td = base.getTableDefForTable(self.conn, "test.adqlgeo")
		self.assertEqual(td.columns[0].name, "row_id")
		table.drop()
		self.assertRaisesWithMsg(
			base.NotFoundError,
			"table 'test.adqlgeo' could not be located in published tables",
			base.getTableDefForTable,
			(self.conn, "test.adqlgeo"))

	def testOnDiskViewMatch(self):
		rd = base.parseFromString(rscdesc.RD, r"""<resource schema="test">
			<table id="aview" namePath="//tap#tables" onDisk="True">
				<column original="table_name"/>
				<column original="utype" name="table_utype"/>
				<viewStatement>
					create view \qName as (SELECT
						table_name,
						utype as table_utype
						from tap_schema.tables)
				</viewStatement></table></resource>""")
		td = rd.getById("aview")
		table = rsc.TableForDef(td, connection=self.conn, nometa=True, create=True)
		try:
			table.ensureOnDiskMatches() # will raise an exception if the test fails
		finally:
			table.drop()

	def testOnDiskHiddenMatchAndDelimited(self):
		rd = base.parseFromString(rscdesc.RD, r"""<resource schema="test">
			<meta>
				description: Forget it
				creationDate: 2021-01-10T10:00:00
				subject: galaxies
			</meta>
			<table id="withhidden" onDisk="True" adql="True">
				<column name="quoted/colA"/>
				<column name="colB" hidden="True"/>
			</table></resource>""")
		td = rd.getById("withhidden")
		tap.publishToTAP(rd, self.conn)
		table = rsc.TableForDef(td, connection=self.conn, nometa=True, create=True)
		self.conn.commit()

		try:
			self.assertEqual([],
				list(validation.validateTable(table.tableDef, True)))
		finally:
			tap.unpublishFromTAP(rd, self.conn)
			table.drop()
			self.conn.commit()

	def testWithMethodIndex(self):
		rd = base.parseFromString(rscdesc.RD, r"""<resource schema="test">
			<table id="hasuri" onDisk="True">
				<index name="gobble" columns="uri">uri text_pattern_ops</index>
				<column name="uri" type="text"/>
			</table></resource>""")
		try:
			table = rsc.TableForDef(rd.getById("hasuri"),
				connection=self.conn, nometa=True, create=True)
			table.importFinished(0)
			self.assertTrue(
				table.hasIndex(table.tableDef.getQName(),
				"hasuri_gobble"))
		finally:
			table.drop()

	def testWithExpressionIndex(self):
		rd = base.parseFromString(rscdesc.RD, r"""<resource schema="test">
			<table id="hasuri" onDisk="True">
				<index name="gobble" columns="path">('http://foo' || path)</index>
				<column name="path" type="text"/>
			</table></resource>""")
		try:
			table = rsc.TableForDef(rd.getById("hasuri"),
				connection=self.conn, nometa=True, create=True)
			table.importFinished(0)
			self.assertTrue(
				table.hasIndex(table.tableDef.getQName(),
				"hasuri_gobble"))
		finally:
			table.drop()


class DBTableQueryTest(tresc.TestWithDBConnection):
	def setUp(self):
		tresc.TestWithDBConnection.setUp(self)
		self.rd = base.parseFromString(rscdesc.RD,
			'<resource schema="testing">'
			'<table id="xy" onDisk="True">'
			'<column name="x" type="integer" required="True"/>'
			'<column name="y" type="text"/></table>'
			'<data id="xyData"><dictlistGrammar/><table original="xy"/>'
			'<rowmaker id="d_xy" idmaps="x,y"/>'
			'<make table="xy" rowmaker="d_xy"/></data>'
			'</resource>')
		self.rd.sourceId = "testing"
		self.data = rsc.makeData(self.rd.getById("xyData"),
			forceSource=[{"x": 9, "y": ""}]+
				[{"x": i, "y": "x"*(9-i)} for i in range(10)],
				connection=self.conn)

	def testPlainQuery(self):
		resdef = svcs.OutputTableDef.fromTableDef(
			self.rd.getTableDefById("xy"), None)
		res = self.data.tables["xy"].getTableForQuery(resdef, "", {})
		self.assertEqual(len(res.rows), 11)
	
	def testDistinctQuery(self):
		resdef = svcs.OutputTableDef.fromTableDef(
			self.rd.getTableDefById("xy"), None)
		res = self.data.tables["xy"].getTableForQuery(resdef, "", {},
			distinct=True)
		self.assertEqual(len(res.rows), 10)

	def testWithLimit(self):
		resdef = svcs.OutputTableDef.fromTableDef(
			self.rd.getTableDefById("xy"), None)
		res = self.data.tables["xy"].getTableForQuery(resdef, "", {},
			limits=("ORDER BY y LIMIT %(limit_)s", {"limit_": 4}))
		self.assertEqual(len(res.rows), 4)
		self.assertEqual(res.rows[-1], {'y': 'xx', 'x': 7})

	def testWithWhere(self):
		resdef = svcs.OutputTableDef.fromTableDef(
			self.rd.getTableDefById("xy"), None)
		res = self.data.tables["xy"].getTableForQuery(resdef,
			"x=%(x)s", {"x":9})
		self.assertEqual(len(res.rows), 2)


class FixupTest(tresc.TestWithDBConnection):
	def testInvalidFixup(self):
		try:
			base.parseFromString(rscdef.TableDef,
			'<table id="test"><column name="ab" fixup="9m+5s"/></table>')
		except base.BadCode as ex:
			self.assertTrue(
				'At IO:\'<table id="test"><column name="ab" fixup="9m+5s"/></t...\', (1, 50):'
				' Bad source code in' in str(ex),
				"Unexcepted error message: %s"%ex)
			return
		self.fail("Exception not raised")
	
	def testSimpleFixup(self):
		td = base.parseFromString(rscdef.TableDef,
			'<table id="test" onDisk="True" temporary="True">'
			'<column name="ab" type="text" fixup="\'ab\'+___"/></table>')
		t = rsc.TableForDef(td, rows=[{"ab": "xy"}, {"ab": "zz"}],
			connection=self.conn, create=True)
		self.assertEqual(
			list(t.iterQuery(svcs.OutputTableDef.fromTableDef(td, None), "")),
			[{'ab': 'abxy'}, {'ab': 'abzz'}])

	def testMultiFixup(self):
		td = base.parseFromString(rscdef.TableDef,
			'<table id="testMulti" onDisk="True" temporary="True">'
			'<column name="ab" type="date"'
			' fixup="___+datetime.timedelta(days=1)"/>'
			'<column name="x" type="integer" fixup="___-2" required="True"/>'
			'<column name="t" type="text" fixup="___ or \'\\test\'"/>'
			'</table>')
		t = rsc.TableForDef(td, rows=[
			{"ab": datetime.date(2002, 2, 2), "x": 14, 't': "ab"},
			{"ab": datetime.date(2002, 2, 3), "x": 15, 't': None}],
			connection=self.conn, create=True)
		self.assertEqual(
			list(t.iterQuery(svcs.OutputTableDef.fromTableDef(td, None), "")), [
				{'x': 12, 'ab': datetime.date(2002, 2, 3),
					't': 'ab'},
				{'x': 13, 'ab': datetime.date(2002, 2, 4),
					't': 'test macro expansion'}])


class STCTest(testhelpers.VerboseTest):
	"""tests for various aspects of STC handling.
	"""
	def testSimpleSTC(self):
		td = base.parseFromString(rscdef.TableDef,
			'<table id="simple">'
			'  <stc>Position ICRS "ra" "dec"</stc>'
			'  <column name="ra" unit="deg"/>'
			'  <column name="dec" unit="deg"/>'
			'  <column name="mag" unit="mag"/>'
			'</table>')
		self.assertEqual(td.getColumnByName("ra").stc.sys.spaceFrame.refFrame,
			"ICRS")
		self.assertEqual(td.getColumnByName("mag").stc, None)

	def testErrorCircle(self):
		td = base.parseFromString(rscdef.TableDef,
			'<table id="errors">'
			'  <stc>Position ICRS 20 20 Error "e_pos" "e_pos"</stc>'
			'  <column name="e_pos"/>'
			'</table>')
		posSTC = td.getColumnByName("e_pos").stc
		self.assertTrue(isinstance(posSTC.place.error, dm.RadiusWiggle))

	def testComplexSTC(self):
		td = base.parseFromString(rscdef.TableDef,
			'<table id="complex">'
			'  <stc>TimeInterval TT "start" "end" Position ICRS "ra" "dec"'
			'    Error "e_ra" "e_dec"</stc>'
			'  <stc>PositionInterval FK5 "raMin" "decMin"</stc>'
			'  <column name="ra" unit="deg"/>'
			'  <column name="dec" unit="deg"/>'
			'  <column name="e_ra" unit="deg"/>'
			'  <column name="e_dec" unit="deg"/>'
			'  <column name="start" type="timestamp"/>'
			'  <column name="end" type="timestamp"/>'
			'  <column name="raMin" unit="deg"/>'
			'  <column name="decMin" unit="deg"/>'
			'  <column name="mag" unit="mag"/>'
			'</table>')
		bigSTC = td.getColumnByName("ra").stc
		self.assertEqual(td.getColumnByName("ra").stc.sys.spaceFrame.refFrame, "ICRS")
		self.assertEqual(td.getColumnByName("mag").stc, None)
		self.assertEqual(td.getColumnByName("start").stc, bigSTC)
		self.assertEqual(td.getColumnByName("end").stc, bigSTC)
		self.assertEqual(td.getColumnByName("start").stc.sys.timeFrame.timeScale,
			"TT")
		self.assertEqual(td.getColumnByName("raMin").stc.sys.spaceFrame.refFrame,
			"FK5")

	def testGeometrySTC(self):
		td = base.parseFromString(rscdef.TableDef,
			'<table id="geo">'
			'  <stc>Box ICRS [bbox]</stc>'
			'  <column name="bbox" type="box"/>'
			'</table>')
		self.assertEqual("ICRS",
			td.getColumnByName("bbox").stc.sys.spaceFrame.refFrame),
	
	def testCopying(self):
		td = base.parseFromString(rscdef.TableDef,
			'<table id="geo">'
			'  <stc>Box ICRS [bbox]</stc>'
			'  <column name="bbox" type="box"/>'
			'</table>')
		tdc = td.copy(None)
		self.assertEqual(tdc.getColumnByName("bbox").stc.sys.spaceFrame.refFrame,
			"ICRS")
		self.assertTrue(tdc.getColumnByName("bbox").stc is
			td.getColumnByName("bbox").stc)


class _ParamTD(testhelpers.TestResource):
	def make(self, ignored):
		return base.parseFromString(rscdef.TableDef,
			'<table id="u"><param name="i" type="integer"/>'
			'<param name="d" type="timestamp">'
			'2011-11-11T11:11:11</param>'
			'<param name="s" type="text"/></table>')


class ParamTest(testhelpers.VerboseTest):
	resources = [("td", _ParamTD())]

	def testPlain(self):
		table = rsc.TableForDef(self.td)
		self.assertEqual(table.getParam("i"), None)
		self.assertEqual(table.getParam("d"), datetime.datetime(
			2011, 11, 11, 11, 11, 11))
	
	def testNoClobber(self):
		table = rsc.TableForDef(self.td)
		table.setParam("i", 10)
		self.assertEqual(table.getParam("i"), 10)
		table2 = rsc.TableForDef(self.td)
		self.assertEqual(table2.getParam("i"), None)
	
	def testParamCons(self):
		table = rsc.TableForDef(self.td, params={
				"i": 10, "d": "2010-10-10T10:10:10"})
		self.assertEqual(table.getParam("i"), 10)
		self.assertEqual(table.getParam("d"), datetime.datetime(
			2010, 10, 10, 10, 10, 10))
	
	def testSetParam(self):
		table = rsc.TableForDef(self.td)
		table.setParam("i", 10)
		table.setParam("d", "2010-10-10T10:10:10")
		self.assertEqual(table.getParam("i"), 10)
		self.assertEqual(table.getParam("d"), datetime.datetime(
			2010, 10, 10, 10, 10, 10))

	def testSetParamFail(self):
		table = rsc.TableForDef(self.td)
		self.assertRaisesWithMsg(base.NotFoundError,
		"column 'doric' could not be located in TableDef u",
		table.setParam,
		("doric", 10))

	def testMacro(self):
		table = rsc.TableForDef(self.td)
		table.setParam("s", r"\metaString{publisher}")
		self.assertEqual(table.getParam("s"),
			base.getMetaText(table, "publisher"))


class QueryTableTest(testhelpers.VerboseTest):
	resources = [
		("basetable", tresc.csTestTable),
		("conn", tresc.dbConnection),
		("ssaTable", tresc.ssaTestTable)]

	def testBasic(self):
		table = rsc.QueryTable(self.basetable.tableDef,
			"SELECT * FROM %s"%self.basetable.tableDef.getQName(),
			base.getDBConnection("trustedquery"), autoClose=True)
		rows = list(table)
		self.assertTrue(isinstance(rows[0], dict))

	def testFromColumns(self):
		with base.getWritableTableConn() as conn:
			table = rsc.QueryTable.fromColumns(
				[self.basetable.tableDef.getColumnByName("alpha"),
					{"name": "mag2", "ucd": "phot.mag;times.two"}],
				"SELECT alpha, mag*2 FROM %s"%self.basetable.tableDef.getQName(),
				connection=conn)
			self.assertEqual(
				list(table),
				[{'alpha': 10.0, 'mag2': 28.0}])

	def testRepeatedIteration(self):
		table = rsc.QueryTable(self.basetable.tableDef,
			"SELECT * FROM %s"%self.basetable.tableDef.getQName(),
			connection=self.conn)
		_ = list(table) #noflake: result doesn't matter for this test
		self.assertRaisesWithMsg(base.ReportableError,
			"QueryTable already exhausted.",
			list,
			(table,))

	def testRefusesRows(self):
		self.assertRaisesWithMsg(base.Error,
			"QueryTables cannot be constructed with rows.",
			rsc.QueryTable,
			(None, "", None),
			rows=[])

	def testCasePreserved(self):
		td = self.ssaTable.tableDef
		table = rsc.QueryTable.fromColumns([td.getColumnByName("ssa_pubDID")],
			"SELECT ssa_pubdid"
			" FROM %s LIMIT 1"%td.getQName(),
			connection=self.conn)
		self.assertEqual(
			[c.name for c in table.tableDef],
			["ssa_pubDID"])


class RAMFeedingTest(testhelpers.VerboseTest):
	def testWorks(self):
		td = base.parseFromString(rscdef.TableDef,
			'<table><column name="x" type="integer" required="True"/></table>')
		instance = rsc.TableForDef(td)
		with instance.getFeeder() as f:
			f.add({'x': 1})
			f.add({'x': 2})
		self.assertEqual(len(instance.rows), 2)
		self.assertEqual(instance.rows[1]['x'], 2)

	def testExceptionRaised(self):
		td = base.parseFromString(rscdef.TableDef,
			'<table><column name="x" type="integer" required="True"/></table>')
		instance = rsc.TableForDef(td)

		def failInFeed():
			with instance.getFeeder():
				raise OverflowError()

		self.assertRaises(OverflowError, failInFeed)

	def testInactiveRaises(self):
		td = base.parseFromString(rscdef.TableDef,
			'<table><column name="x" type="integer" required="True"/></table>')
		instance = rsc.TableForDef(td)
		f = instance.getFeeder()
		with f:
			f.add({'x': 1})
		self.assertRaises(base.DataError, f.add, {'x': 1})


class DBFeedingTest(tresc.TestWithDBConnection):
	def testInactiveRaises(self):
		td = base.parseFromString(rscdef.TableDef,
			'<table id="bla" temporary="true" onDisk="true">'
				'<column name="x" type="integer" required="True"/></table>')
		instance = rsc.TableForDef(td, connection=self.conn, create=True)
		try:
			f = instance.getFeeder()
			with f:
				f.add({'x': 1})
			self.assertRaises(base.DataError, f.add, {'x': 1})
		finally:
			self.conn.rollback()

	def testFlushWorks(self):
		td = base.parseFromString(rscdef.TableDef,
			'<table id="bla" temporary="true" onDisk="true">'
				'<column name="x" type="integer" required="True"/></table>')
		instance = rsc.TableForDef(td, connection=self.conn, create=True)
		try:
			f = instance.getFeeder()
			with f:
				f.add({'x': 1})
				self.assertEqual(len(f.batchCache), 1)
				self.assertEqual(list(instance.iterQuery(td, "")), [])
			self.assertEqual(list(instance.iterQuery(td, "")), [{'x': 1}])
			with f:
				f.add({'x': 2})
				f.flush()
				self.assertEqual(list(instance.iterQuery(td, "")),
					[{'x': 1}, {'x': 2}])
		finally:
			self.conn.rollback()

	def testResetWorks(self):
		td = base.parseFromString(rscdef.TableDef,
			'<table id="bla" temporary="true" onDisk="true">'
				'<column name="x" type="integer" required="True"/></table>')
		instance = rsc.TableForDef(td, connection=self.conn, create=True)
		try:
			f = instance.getFeeder()
			with f:
				f.add({'x': 1})
				f.reset()
			self.assertEqual(list(instance.iterQuery(td, "")), [])
		finally:
			self.conn.rollback()


class ViewDefinitionTest(testhelpers.VerboseTest):
	def testIndexInheritance(self):
		rd = base.parseFromString(rscdesc.RD,
			r"""<resource schema="test">
				<table id="hastwoind" onDisk="True">
					<index columns="cola"/>
					<index columns="colb"/>
					<column name="cola"/>
					<column name="colb"/>
				</table>

				<table id="myview" onDisk="True">
					<column original="hastwoind.colb"/>
					<column original="data/cores#conecat.ra"/>
					<column original="data/cores#conecat.dec"/>

					<FEED source="//procs#declare-indexes-from"
						sourceTables="hastwoind data/cores#conecat"/>

					<viewStatement>
						create view \qName as (
							select \colNames from
								\schema.hastwoind, test.conecat)
					</viewStatement>
				</table>
			</resource>""")
		td = rd.getById("myview")
		indexMeta = [(ind.columns, ind.kind) for ind in td.indices]
		indexMeta.sort()
		self.assertEqual(indexMeta,
			[(['colb'], 'straight'), (['ra', 'dec'], 'q3c')])
		

class AutoMetaTest(testhelpers.VerboseTest):
	def testReferenceURL(self):
		self.assertEqual(
			base.getMetaText(
				base.resolveCrossId("data/test#typesTable"),
				"referenceURL"),
			"http://localhost:8080/tableinfo/test.typesTable")

	def testShortName(self):
		self.assertEqual(
			base.getMetaText(
				base.resolveCrossId("data/test#typesTable"),
				"shortName"),
			"test.typesTable")

	def testHowToCiteLink(self):
		self.assertEqual(
			base.getMetaText(
				base.resolveCrossId("data/test#typesTable"),
				"howtociteLink"),
			"http://localhost:8080/tableinfo/test.typesTable#ti-citing")
	

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