"""
Simple tests for TAP.

All these tests really stink because TAP isn't really a good match for the
basically stateless unit tests that are executed in an arbitrary order.

There's more TAP/UWS related tests in test_tap.py; these require a
running reactor and are based on trial.
"""

#c Copyright 2008-2021, 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
import io
import functools
import os
import time
import threading
import traceback

from gavo.helpers import testhelpers

from gavo import base
from gavo import rscdesc  # uws needs getRD
from gavo import svcs
from gavo import utils
from gavo import votable
from gavo.helpers import testtricks
from gavo.protocols import adqlglue
from gavo.protocols import tap
from gavo.protocols import taprunner
from gavo.protocols import uws
from gavo.protocols import uwsactions
from gavo.registry import capabilities
from gavo.web import vosi

import adqltest
import tresc

@functools.lru_cache(None)
def _getTAPService():
	return base.caches.getRD("//tap").getById("run")

@functools.lru_cache(None)
def _getTAPParGrammar():
	return _getTAPService().getContextGrammarFor("sync", None)


class TAPFakeRequest(testhelpers.FakeRequest):
  def __init__(self, *posargs, **kwargs):
  	testhelpers.FakeRequest.__init__(self, *posargs, **kwargs)
  	uws.prepareRequest(self, _getTAPService())
  	self.asyncGrammar = _getTAPParGrammar()

def _makeATAPJob(pars):
	"""returns a job id for a TAP job described by pars.

	pars needs to be like request.strargs.
	"""
	return tap.WORKER_SYSTEM.getNewJobId(
		parameters=_getTAPParGrammar().parseStrargs(pars, ensureRequired=False))


class TestVersionParsing(testhelpers.VerboseTest):
	def testSupportedLang(self):
		self.assertEqual(
			taprunner._assertSupportedLanguage(None, "ADQL"), None)

	def testSupportedLangVersion(self):
		self.assertEqual(
			taprunner._assertSupportedLanguage(None, "ADQL-2.0"), None)

	def testAnotherSupportedLangVersion(self):
		self.assertEqual(
			taprunner._assertSupportedLanguage(None, "ADQL-2.1"), None)

	def testUnsupportedLang(self):
		self.assertRaisesWithMsg(
			uws.UWSError,
			"This service does not support the query language ShittyQL",
			taprunner._assertSupportedLanguage,
			(None, "ShittyQL"))

	def testUnsupportedVersion(self):
		self.assertRaisesWithMsg(
			uws.UWSError,
			"This service does not support version 30.12-pre_foo of the query"
			" language ADQL (but some other version; see capabilities).",
			taprunner._assertSupportedLanguage,
			(None, "ADQL-30.12-pre_foo"))


class TestResponseformat(testhelpers.VerboseTest):
	def testRejection(self):
		self.assertRaisesWithMsg(base.ValidationError,
			"Field responseformat: 'xls' is not a valid value for responseformat",
			_getTAPService().run, ("sync", {
					"query": ["ignore"], "lang": ["ADQL"], 
					"responseformat": ["xls"]}
				, None))
	
	def testAlias(self):
		args = {"lang": ["ADQL"], "responseformat": ["votable/td"],
				"query": ["SELECT TOP 1 * FROM tap_schema.tables"]}
		f, mediaType = base.resolveCrossId("//tap#run").run("sync", 
			_getTAPParGrammar().parseStrargs(args),
			svcs.QueryMeta(args))
		self.assertEqual(mediaType, 
			"application/x-votable+xml;serialization=TABLEDATA")
		self.assertEqual(f.read()[-50:], 
			b"R></TABLEDATA></DATA></TABLE></RESOURCE></VOTABLE>")

	def testPreservedMediaType(self):
		args = {"lang": ["ADQL"], "responseformat": ["text/xml"],
				"query": ["SELECT TOP 1 * FROM tap_schema.tables"]}
		f, mediaType = base.resolveCrossId("//tap#run").run("sync", 
			_getTAPParGrammar().parseStrargs(args),
			svcs.QueryMeta(args))
		self.assertEqual(mediaType, "text/xml")
		# I take it that text/xml ought to be formattable with XSLT, and
		# hence we want tabledata.
		self.assertEqual(f.read()[-50:], 
			b"R></TABLEDATA></DATA></TABLE></RESOURCE></VOTABLE>")


class PlainJobCreationTest(testhelpers.VerboseTest):
	"""tests for working job creation and destruction.
	"""
	resources = [("conn", tresc.dbConnection)]

	def _createJob(self):
		# pass in an invalid key so we can see that it's swallowed.
		return _makeATAPJob({"foo": "bar"})

	def _deleteJob(self, jobId):
		return tap.WORKER_SYSTEM.destroy(jobId)

	def _assertJobCreated(self, jobId):
		res = list(self.conn.query(
			"SELECT jobId FROM tap_schema.tapjobs WHERE"
				" jobId=%(jobId)s", locals()))
		self.assertEqual(len(res), 1)
		job = tap.WORKER_SYSTEM.getJob(jobId)
		self.assertFalse("foo" in job.parameters)
		self.assertTrue(os.path.exists(job.getWD()))

	def _assertJobDeleted(self, jobId):
		res = list(self.conn.query(
			"SELECT destructiontime FROM tap_schema.tapjobs"
			" WHERE jobId=%(jobId)s", locals()))
		self.assertEqual(len(res), 0)
		self.assertRaises(base.NotFoundError, tap.WORKER_SYSTEM.getJob, jobId)
		self.assertFalse(
			os.path.exists(
				os.path.join(base.getConfig("uwsWD"), jobId)))

	def _assertUnqueueable(self, jobId):
		# we're missing all the relevant parameters, and hence the
		# this needs to blow up when we try to queue the job
		tap.WORKER_SYSTEM.changeToPhase(jobId, uws.QUEUED)
		job = tap.WORKER_SYSTEM.getJob(jobId)
		self.assertEqual(job.phase, uws.ERROR)
		self.assertEqual(job.error["msg"], 
			"Field lang: Required parameter 'lang' missing.")

	def testBigAndUgly(self):
		# this may leave its job behind if the assertions fail.  They'll
		# eventually be garbage-collected by DaCHS, so I don't worry too much.
		jobId = self._createJob()
		self._assertJobCreated(jobId)
		self._assertUnqueueable(jobId)
		self._deleteJob(jobId)
		self._assertJobDeleted(jobId)


class _UWSJobResource(testhelpers.TestResource):
# just a UWS job.  Don't manipulate it.  Too badly.
	def make(self, ignored):
		return tap.WORKER_SYSTEM.getNewJobId()
	
	def clean(self, jobId):
		tap.WORKER_SYSTEM.destroy(jobId.original)

_uwsJobResource = _UWSJobResource()


class TAPParametersTest(testhelpers.VerboseTest):
	resources = [("jobIdResource", _uwsJobResource)]

	def setUp(self):
		testhelpers.VerboseTest.setUp(self)
		self.jobId = self.jobIdResource.original

	def testRepeatedSetting(self):
		with tap.WORKER_SYSTEM.changeableJob(self.jobId) as job:
			job.setParamsFromDict({"a": ["ob"], "b": ["knapp"]})
			self.assertEqual(job.parameters["a"], ["ob"])
			job.setParamsFromDict({"a": ["ja"], "b": []})
			self.assertEqual(job.parameters["a"], ["ob", "ja"])
			self.assertEqual(job.parameters["b"], ["knapp"])


class UWSResponsesValidTest(testhelpers.VerboseTest, testtricks.XSDTestMixin):
	resources = [("jobIdResource", _uwsJobResource)]

	def setUp(self):
		testhelpers.VerboseTest.setUp(self)
		self.jobId = self.jobIdResource.original

	def testJobRes(self):
		job = tap.WORKER_SYSTEM.getJob(self.jobId)
		self.assertValidates(uwsactions.RootAction().doGET(
			job, TAPFakeRequest()),
			leaveOffending=True)

	def testJobList(self):
		self.assertValidates(uwsactions.getJobList(tap.WORKER_SYSTEM), 
			leaveOffending=True)


class BlockingTest(testhelpers.VerboseTest):
	"""tests for working impicit uws locking.
	"""
	resources = [("jobIdResource", _uwsJobResource)]

	def setUp(self):
		testhelpers.VerboseTest.setUp(self)
		self.jobId = self.jobIdResource.original

	def testIndexDoesNotBlock(self):
		with tap.WORKER_SYSTEM.changeableJob(self.jobId):
			self.assertTrue(b"uws:jobs" in uwsactions.getJobList(tap.WORKER_SYSTEM))

	def testGetPhaseDoesNotBlock(self):
		req = TAPFakeRequest()
		with tap.WORKER_SYSTEM.changeableJob(self.jobId):
			self.assertEqual(
				uwsactions.doJobAction(tap.WORKER_SYSTEM, req, (self.jobId, "phase")),
				"PENDING")

	def testPostPhaseDoesBlock(self):
		req = TAPFakeRequest(args={"PHASE": ["RUN"]})
		req.method = b"POST"
		uwsactions.PhaseAction.timeout = 0.05
		with tap.WORKER_SYSTEM.changeableJob(self.jobId):
			self.assertRaisesWithMsg(base.ReportableError,
				"Could not access the jobs table. This probably means"
				" there is a stale lock on it.  Please notify the service operators.",
				uwsactions.doJobAction,
				(tap.WORKER_SYSTEM, req, (self.jobId, "phase")))


class QueueTest(testhelpers.VerboseTest):
	def testQuote(self):
		now = datetime.datetime.utcnow()
		tick = datetime.timedelta(minutes=15)
		jobs = []
		try:
			jobId = tap.WORKER_SYSTEM.getNewJobId()
			with tap.WORKER_SYSTEM.changeableJob(jobId) as wjob:
				wjob.change(phase=uws.QUEUED, destructionTime=now+10*tick)
			jobs.append(jobId)

			testJob = tap.WORKER_SYSTEM.getJob(jobId)
			# don't fail just because jobs are left in the queue
			baseDelay = testJob.quote-now

			jobId = tap.WORKER_SYSTEM.getNewJobId()
			with tap.WORKER_SYSTEM.changeableJob(jobId) as wjob:
				wjob.change(phase=uws.QUEUED, destructionTime=now+9*tick)
			jobs.append(jobId)
			
			# our quote must now be roughly 10 minutes (as configured in
			# tap.EST_TIME_PER_JOB) later
			self.assertEqual((testJob.quote-now-baseDelay).seconds/10, 
				tap.EST_TIME_PER_JOB.seconds/10)

			jobId = tap.WORKER_SYSTEM.getNewJobId()
			with tap.WORKER_SYSTEM.changeableJob(jobId) as wjob:
				wjob.change(phase=uws.QUEUED, destructionTime=now+11*tick)
			jobs.append(jobId)

			# the new job will run later then our test job, so no change
			# expected
			self.assertEqual((testJob.quote-now-baseDelay).seconds/10, 
				tap.EST_TIME_PER_JOB.seconds/10)

		finally:
			for jobId in jobs:
				tap.WORKER_SYSTEM.destroy(jobId)


class TAPTransitionsTest(testhelpers.VerboseTest):
	def testAbortPending(self):
		jobId = None
		try:
			jobId = tap.WORKER_SYSTEM.getNewJobId(
				parameters={"query": "bogus", "request": "doQuery",
				"LANG": "ADQL-2.0"})
			tap.WORKER_SYSTEM.changeToPhase(jobId, uws.ABORTED)
			self.assertEqual(tap.WORKER_SYSTEM.getJob(jobId).phase, 
				uws.ABORTED)
		finally:
			tap.WORKER_SYSTEM.destroy(jobId)


class UploadSyntaxOKTest(testhelpers.VerboseTest, metaclass=testhelpers.SamplesBasedAutoTest):
	def _runTest(self, sample):
		s, e = sample
		self.assertEqual(tap.parseUploadString(s), e)
	
	samples = [
		('a,b', [('a', 'b'),]),
		('a5_ug,http://knatter?RA=99&DEC=1.54', 
			[('a5_ug', 'http://knatter?RA=99&DEC=1.54'),]),
		('a5_ug,http://knatter?RA=99&DEC=1.54;a,b', 
			[('a5_ug', 'http://knatter?RA=99&DEC=1.54'), ('a', 'b')]),]


class UploadSyntaxNotOKTest(testhelpers.VerboseTest, metaclass=testhelpers.SamplesBasedAutoTest):
	def _runTest(self, sample):
		self.assertRaises(base.ValidationError, tap.parseUploadString,
			sample)
	
	samples = [
		'a,',
		',http://wahn',
		'a,b;',
		'a,b;whacky/name,b',]


class _TAPCapDocument(testhelpers.TestResource):
	def make(self, ignored):
		pub = base.caches.getRD("//tap").getById("run").publications[0]
		doc = vosi.CAP.capabilities[capabilities.getCapabilityElement(pub)
			].render()
		return doc, testhelpers.getXMLTree(doc, debug=False)


class CapabilitiesTest(testhelpers.VerboseTest, testtricks.XSDTestMixin):
	resources = [("dt", _TAPCapDocument())]

	def testValid(self):
		self.assertValidates(self.dt[0], leaveOffending=True)
	
	def testFunctionsSorted(self):
		funcs = [n.text for n in 
			self.dt[1].xpath("//languageFeatures[@type='ivo://ivoa.net/std/"
				"TAPRegExt#features-udf']/feature/form")]
		self.assertEqual(list(sorted(funcs)), funcs)

	def testUploadMethods(self):
		ums = set(n.get("ivo-id") for n in self.dt[1].xpath("//uploadMethod"))
		self.assertEqual(set(ums),
			set("ivo://ivoa.net/std/TAPRegExt#upload-"+s
				for s in ("inline", "http", "https", "ftp")))

	def testType(self):
		self.assertEqual(self.dt[1].xpath("capability[1]")[0].get(
				"{http://www.w3.org/2001/XMLSchema-instance}type"),
			"tr:TableAccess")

	def testInterface(self):
		intfs = self.dt[1].xpath("capability[1]/interface")
		self.assertEqual(len(intfs), 2)
		intf = intfs[0]
		self.assertEqual(
			intf.get("{http://www.w3.org/2001/XMLSchema-instance}type"),
			"vs:ParamHTTP")

	def testAccessURL(self):
		self.assertEqual([n.text for n in 
			self.dt[1].xpath("capability/interface/accessURL")],
			["http://localhost:8080/tap", "http://localhost:8080/tap"])


class TAPSchemaTest(testhelpers.VerboseTest):
	def setUp(self):
		self.jobId = _makeATAPJob({
				"query": ["SELECT TOP 1 * FROM TAP_SCHEMA.tables"],
				"request": ["doQuery"],
				"lang": ["ADQL-2.1"]})
	
	def tearDown(self):
		tap.WORKER_SYSTEM.destroy(self.jobId)

	def testTAP_tables(self):
		taprunner.runSyncTAPJob(self.jobId)
		job = tap.WORKER_SYSTEM.getJob(self.jobId)
		self.assertEqual(job.phase, uws.COMPLETED)
		res = open(job.getResult("result")[0]).read()
		self.assertTrue('<RESOURCE type="results">' in res)


class EvilUploadTest(testhelpers.VerboseTest):
	def _runWithUpload(self, query, uploadContent):
		jobId = _makeATAPJob({
			"query": [query],
			"request": ["doQuery"],
			"lang": ["ADQL"],
			"responseformat": ["votable/td"],
			"upload": [("t1",  "file://evil.vot")]})

		job = tap.WORKER_SYSTEM.getJob(jobId)
		with testhelpers.testFile("evil.vot", uploadContent,
				inDir=job.getWD()):
			try:
				taprunner.runSyncTAPJob(jobId)
				job = tap.WORKER_SYSTEM.getJob(jobId)
				self.assertEqual(job.phase, uws.COMPLETED)
				res = open(job.getResult("result")[0]).read()
				return res
			finally:
				tap.WORKER_SYSTEM.destroy(jobId)

	def testEvilColnameAndXtype(self):
		# because running these tests is expensive, I'm abusing this
		# to also test for fantasy xtype roundtripping.
		result = self._runWithUpload(
			"SELECT * FROM TAP_SCHEMA.tables join TAP_UPLOAD.t1 ON oid=table_index",
				"""<VOTABLE version="1.2" 
					xmlns="http://www.ivoa.net/xml/VOTable/v1.2">
				<RESOURCE>
				<TABLE name="holler" nrows="1">
				<FIELD datatype="long" name="oid" ucd="meta.record;meta.id"
					xtype="magic:oid"/>
				<DATA><TABLEDATA><TR><TD>1</TD></TR></TABLEDATA></DATA>
				</TABLE>
				</RESOURCE>
				</VOTABLE>""")
		self.assertTrue('name="oid"' in result)
		self.assertTrue('xtype="magic:oid"' in result)

	def testArrayWithNaN(self):
		result = self._runWithUpload(
			"SELECT column_index FROM TAP_SCHEMA.columns join TAP_UPLOAD.t1 ON"
				" column_index between intv[1] and intv[2]",
				"""<VOTABLE xmlns="http://www.ivoa.net/xml/VOTable/v1.3">
				<RESOURCE><TABLE>
				<FIELD datatype="float" arraysize="2" name="intv"/>
				<DATA><TABLEDATA>
					<TR><TD>1.5 2.5</TD></TR>
					<TR><TD>4.9 NaN</TD></TR>
					<TR><TD>NaN NaN</TD></TR>
				</TABLEDATA></DATA>
				</TABLE></RESOURCE></VOTABLE>""")
		self.assertTrue("Contains traces from table tap_schema.columns" in result)
		self.assertTrue("<TD>2</TD>" in result)
		self.assertFalse("<TD>5</TD>" in result)


def _getUnparsedQueryResult(query):
	jobId = _makeATAPJob({
			"query": [query],
			"request": ["doQuery"],
			"lang": ["ADQL"]})
	try:
		taprunner.runSyncTAPJob(jobId)
		job = tap.WORKER_SYSTEM.getJob(jobId)
		if job.phase==uws.ERROR:
			raise Exception("Job died with msg %s"%job.error)
		name, mime = job.getResult("result")
		with open(name, "rb") as f:
			res = f.read()
	finally:
		tap.WORKER_SYSTEM.destroy(jobId)
	return res

def _getQueryResult(query):
	# returns a votable.simple result for query.
	vot = _getUnparsedQueryResult(query)
	res = votable.load(io.BytesIO(vot))
	return res


class SimpleRunnerTest(testhelpers.VerboseTest):
	"""tests various taprunner scenarios.
	"""
	resources = [("ds", adqltest.adqlTestTable)]

	def setUp(self):
		testhelpers.VerboseTest.setUp(self)
		self.tableName = self.ds.tables["adql"].tableDef.getQName()

	def testSimpleJob(self):
		jobId = _makeATAPJob({
			"query": ["SELECT * FROM %s"%self.tableName],
			"request": ["doQuery"],
			"lang": ["ADQL"]})
		try:
			job = tap.WORKER_SYSTEM.getJob(jobId)
			self.assertEqual(job.phase, uws.PENDING)
			tap.WORKER_SYSTEM.changeToPhase(jobId, uws.QUEUED, None)
		
			runningPhases = set([uws.QUEUED, uws.UNKNOWN, uws.EXECUTING])
			# let things run, but bail out if nothing happens 
			for i in range(100):
				time.sleep(0.1)
				job.update()
				if job.phase not in runningPhases:
					break
			else:
				raise AssertionError("Job does not finish.  Your machine cannot be"
					" *that* slow?")
			self.assertEqual(job.phase, uws.COMPLETED)
			with open(os.path.join(job.getWD(), 
						job.getResults()[0]["resultName"]), "rb") as f:
				result = f.read()
			self.assertTrue((datetime.datetime.utcnow()-job.endTime).seconds<1)

		finally:
			job = tap.WORKER_SYSTEM.destroy(jobId)
		self.assertTrue(b'xmlns="http://www.ivoa.net/xml/VOTable/' in result)

	def testColumnNames(self):
		table, meta = _getQueryResult(
			'SELECT cos(delta) as frob, alpha as "AlPhA", delta, "delta",'
			' 20+30, 20+30 AS constant, rv as "AS"'
			' from %s'%self.tableName)
		fields = meta.getFields()
		self.assertEqual(fields[0].name, "frob")
		self.assertEqual(fields[0].getDescription(), "A sample Dec --"
			" *TAINTED*: the value was operated on in a way that unit and"
			" ucd may be severely wrong")
		self.assertEqual(fields[1].name, "AlPhA")
		self.assertEqual(fields[2].name, "delta")
		# the next line tests for a documented violation of the spec.
		self.assertEqual(fields[3].name, "delta_")
		self.assertTrue(isinstance(fields[4].name, str))
		self.assertEqual(fields[5].name, "constant")
		self.assertEqual(fields[6].name, "AS")

	def testColumnTypes(self):
		table, meta = _getQueryResult(
			"SELECT rv, point('icrs', alpha, delta), PI() from %s"%self.tableName)
		fields = meta.getFields()
		self.assertEqual(fields[0].datatype, "double")
		self.assertEqual(fields[1].datatype, "double")
		self.assertEqual(fields[1].xtype, "point")
		self.assertEqual(fields[2].datatype, "double")


class _TAPResultTable(testhelpers.TestResource):
	resources = [("ds", adqltest.adqlTestTable)]

	def make(self, deps):
		return testhelpers.getXMLTree(
			_getUnparsedQueryResult(
				"SELECT rv, PI() from %s"%
					deps["ds"].tables["adql"].tableDef.getQName()), debug=False)


class VOTableMetaTest(testhelpers.VerboseTest):
	resources = [("tree", _TAPResultTable())]

	def testTranslatedQueryPresent(self):
		self.assertEqualIgnoringAliases(
			self.tree.xpath("//INFO[@name='sql_query']")[0].get("value"),
			"SELECT rv, PI() ASWHATEVER FROM test.adql LIMIT 20000")

	def testADQLPresent(self):
		self.assertEqualIgnoringAliases(
			self.tree.xpath("//INFO[@name='query']")[0].get("value"),
			"SELECT rv, PI() from test.adql")

	def testJoinedTablesMentioned(self):
		self.assertEqual(self.tree.xpath("//INFO[@name='src_res']")[0].get("value"),
			"Contains traces from resource data/test")
		self.assertEqual(self.tree.xpath(
			"//INFO[@name='src_table']")[0].get("value"),
			"Contains traces from table test.adql")
	
	def testCopyrightsMentioned(self):
		self.assertEqual(self.tree.xpath(
			"//INFO[@name='copyright']")[0].get("value"),
			"data/test copyright or license")
		self.assertEqual(self.tree.xpath("//INFO[@name='copyright']")[0].text,
			"Everything in here is pure fantasy (distributed under the GNU GPL)")

	def testSourcesMentioned(self):
		self.assertEqual(
			self.tree.xpath("//INFO[@ucd='meta.bib.bibcode']")[0].get("value"),
			"1635QB41.G135......")

	def testSourcesUniqued(self):
		self.assertEqual(len(self.tree.xpath("//INFO[@ucd='meta.bib.bibcode']")), 1)

	def testCitationInfo(self):
		info = self.tree.xpath("//INFO[@ucd='meta.bib']")[0]
		self.assertEqual(info.get("value"), 
			"http://localhost:8080/tableinfo/test.adql")
		self.assertTrue("For advice " in info.text)

	def testAllCitationInfo(self):
		self.assertEqual(len(self.tree.xpath("//INFO[@name='citation']")), 2)

	def testInfoCarriedOver(self):
		self.assertEqual(self.tree.xpath("//INFO[@name='ORIGIN']")[0].get("value"),
			"test.rd")


class _TAPJoinResultTable(testhelpers.TestResource):
	resources = [("ds", adqltest.adqlTestTable), 
		("ssa", tresc.ssaTestTable)]

	def make(self, deps):
		return testhelpers.getXMLTree(
			_getUnparsedQueryResult(
				"SELECT delta, ssa_pubdid"
				" from test.adql join test.hcdtest on (alpha=ssa_length)"),
			debug=False)


class JoinMetaTest(testhelpers.VerboseTest):
	resources = [("tree", _TAPJoinResultTable())]

	def testAllSourcesPresent(self):
		self.assertEqual(set(e.get("value") 
			for e in self.tree.xpath("//INFO[@ucd='meta.bib.bibcode']")),
			set(["1635QB41.G135......", "2015ivoa.spec.0617D"]))

	def testAllCitationsPresent(self):
		self.assertEqual(set(e.get("value") 
			for e in self.tree.xpath("//INFO[@name='citation' and @ucd='meta.bib']")),
			set(['http://localhost:8080/tableinfo/test.adql',
				'http://localhost:8080/tableinfo/test.hcdtest']))

	def testDatalinkServiceMentioned(self):
		svcDefinition = self.tree.xpath("//RESOURCE[@type='meta']")
		self.assertEqual(len(svcDefinition), 1)
		idid = svcDefinition[0].xpath("GROUP/PARAM[@name='ID']")[0].get("ref")
		self.assertEqual(
			self.tree.xpath("//FIELD[@ID='%s']"%idid)[0].get("name"), 
			"ssa_pubDID")
	
	def testDatalinkServiceSkippedWithoutColumn(self):
		tree = testhelpers.getXMLTree(
			_getUnparsedQueryResult(
				"SELECT delta"
				" from test.adql join test.hcdtest on (alpha=ssa_length)"),
			debug=False)
		self.assertEqual(len(tree.xpath("//RESOURCE[@type='meta']")), 0)


class TAPTableMetaTest(testhelpers.VerboseTest):
	resources = [("conn", tresc.dbConnection), ("ot", tresc.obscoreTable),
		("ssa", tresc.ssaTestTable)]

	def testSingleSourceTableMetadataCopied(self):
		res = adqlglue.runTAPQuery(
			"SELECT TOP 1 * FROM ivoa.obscore", 1, self.conn, [], 1,
			autoClose=False)
		list(res) # exhaust the cursor
		self.assertEqual(
			base.getMetaText(res, "description"),
			"The IVOA-defined obscore table, containing generic metadata for\n"
			"datasets within this datacenter.")
	
	def testSingleSourceTableParamsCopied(self):
		res = adqlglue.runTAPQuery(
			"SELECT TOP 1 * FROM test.hcdtest", 1, self.conn, [], 1,
			autoClose=False)
		list(res) # exhaust the cursor
		self.assertEqual(res.getParam("ssa_model"), "Spectrum-1.0")
		self.assertEqual(res.getParam("ssa_specres"), 1e-10)

	def testNoMetadataOnComplexQuery(self):
		res = adqlglue.runTAPQuery(
			"SELECT TOP 1 * FROM (SELECT * from ivoa.obscore) as q", 
				1, self.conn, [], 1, autoClose=False)
		self.assertEqual(
			base.getMetaText(res, "description"), None)
		list(res)

	def testNoMetadataOnJoin(self):
		res = adqlglue.runTAPQuery(
			"SELECT TOP 1 * FROM ivoa.obscore as a join ivoa.obscore as b"
				" ON (a.obs_collection=b.dataproduct_subtype)", 
				1, self.conn, [], 1, autoClose=False)
		self.assertEqual(
			base.getMetaText(res, "description"), None)
		list(res)

	def testCasePreserved(self):
		res = adqlglue.runTAPQuery(
			"SELECT TOP 1 ssa_pubDID FROM %s"%self.ssa.tableDef.getQName(), 
			1, self.conn, [], 1, autoClose=False)
		list(res)
		self.assertEqual(res.tableDef.columns[0].name, "ssa_pubDID")


class JobMetaTest(testhelpers.VerboseTest):
	def setUp(self):
		self.jobId = tap.WORKER_SYSTEM.getNewJobId()
	
	def tearDown(self):
		tap.WORKER_SYSTEM.destroy(self.jobId)

	def _postRedirectCheck(self, req, segments, method=b"POST"):
		req.method = method
		try:
			return uwsactions.doJobAction(tap.WORKER_SYSTEM, req,
				segments=(self.jobId,)+segments)
		except svcs.SeeOther:
			pass # that's the 303 expected
		else:
			self.fail("POSTing didn't redirect")

	def testExecD(self):
		res = uwsactions.doJobAction(tap.WORKER_SYSTEM, 
			TAPFakeRequest(),
			segments=(self.jobId, "executionduration"))
		self.assertEqual(res, "3600")
	
	def testSetExecD(self):
		self._postRedirectCheck(
			TAPFakeRequest(args={"EXECUTIONDURATION": ["300"]}),
			("executionduration",))
		res = uwsactions.doJobAction(tap.WORKER_SYSTEM, 
			TAPFakeRequest(),
			segments=(self.jobId, "executionduration"))
		self.assertEqual(res, "300")

	def testDestruction(self):
		self._postRedirectCheck(
			TAPFakeRequest(
				args={"DESTRuction": ["2000-10-10T10:12:13"]}),
			("destruction",))
		res = uwsactions.doJobAction(tap.WORKER_SYSTEM, 
			TAPFakeRequest(),
			segments=(self.jobId, "destruction"))
		self.assertEqual(res, "2000-10-10T10:12:13Z")
	
	def testQuote(self):
		res = uwsactions.doJobAction(tap.WORKER_SYSTEM, 
			TAPFakeRequest(),
			segments=(self.jobId, "quote"))
		# just make sure it's an iso date
		utils.parseISODT(res)

	def testNullError(self):
		res = uwsactions.doJobAction(tap.WORKER_SYSTEM, TAPFakeRequest(),
			segments=(self.jobId, "error"))
		self.assertEqual(res, b"")

	def testParametersPostAll(self):
		self._postRedirectCheck(
			TAPFakeRequest(args={"QUERY": ["Nicenice"]}),
			("parameters",))

		res = uwsactions.doJobAction(tap.WORKER_SYSTEM, TAPFakeRequest(),
			segments=(self.jobId, "parameters"))
		self.assertTrue(b"Nicenice" in res.render())

	def testOwner(self):
		res = uwsactions.doJobAction(tap.WORKER_SYSTEM, TAPFakeRequest(),
			segments=(self.jobId, "owner"))
		self.assertEqual(res, b"")


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

	def testSchemaPublished(self):
		res = list(self.conn.queryToDicts(
			"select * from tap_schema.schemas where schema_name='test'"))[0]
		self.assertEqual(res["description"],
			"Sundry test tables and other fancy stuff")
		self.assertEqual(res["utype"],
			"http://www.g-vo.org/DaCHS")
		self.assertEqual(res["schema_index"], None)

	def testColumnsPublished(self):
		res = list(self.conn.query(
			"select column_name, unit, datatype, arraysize, description"
			" from tap_schema.columns where table_name='test.csdata'"))
		self.assertTrue(("alpha", "deg", "float", None, "A sample RA") in res)
		self.assertEqual(len(res), 5)

	def testModelsSupportedPresent(self):
		res = list(self.conn.query(
			"select * from tap_schema.supportedmodels where sourcerd='data/test'"))
		self.assertEqual(len(res), 2)
		self.assertEqual(set(r[1] for r in res),
			set(['Fantasy-1.0', 'Fantasy-1.1']))

	def testSTCGroupPresent(self):
		res = set(list(self.conn.query(
			"select * from tap_schema.groups")))
		self.assertTrue(
			('test.adql', 'alpha', 'col:weird.reason', 
				'weird_columns', 'col:weird.name', "data/test") in res)
		self.assertTrue(
			('test.adql', 'rv', None, 'nice_columns', 'col:nice.name', "data/test")
				in res)


class _RenderedTAPExample(testhelpers.TestResource):
	def make(self, ignored):
		from gavo.web import examplesrender

		rd = base.parseFromString(rscdesc.RD,
			"""<resource schema="test"><meta name="_example"
				title="Query against boolean columns">
				Some tables this
				service exposes to TAP -- e.g., :taptable:`amanda.nucand`. 
				To query those, you cannot say ``WHERE boolCol`` but have to compare
				``'True'`` or ``'False'`` as in

				.. tapquery::

					SELECT * 
					  FROM amanda.nucand 
					WHERE atmonusubset='True' and foo&lt;20
					
				(nothing else will work).</meta></resource>""")
		return testhelpers.getXMLTree(
			examplesrender._Example(
				rd.getMeta("_example"))._getTranslatedHTML(), debug=False)


class TAPExampleTest(testhelpers.VerboseTest):
	resources = [("tree", _RenderedTAPExample())]

	def testContainerDeclared(self):
		root = self.tree.xpath('/div[@typeof="example"]')[0]
		self.assertEqual(root.get("id"), "Queryagainstbooleancolumns")
		self.assertEqual(root.get("resource"), "#Queryagainstbooleancolumns")

	def testTitleMarkedUp(self):
		title = self.tree.xpath("h2")[0]
		self.assertEqual(title.get("property"), "name")
		self.assertEqual(title.text, "Query against boolean columns")

	def testPlainText(self):
		para = self.tree.xpath("p")[0]
		self.assertTrue(para.text.startswith("Some tables"))
	
	def testTableLink(self):
		tableAnchor = self.tree.xpath("//*[@property='table']")[0]
		self.assertEqual(len(tableAnchor), 0) # no children
		self.assertEqual(tableAnchor.text, "amanda.nucand")
		tableLink = self.tree.xpath("//a")[0]
		self.assertEqual(tableLink.get("href"), "/tableinfo/amanda.nucand")

	def testQueryPresent(self):
		queryEl = self.tree.xpath("//*[@property='query']")[0]
		self.assertEqual(queryEl.text.strip(),
			"SELECT *\n  FROM amanda.nucand\nWHERE atmonusubset='True' and foo<20")
		self.assertEqual(queryEl.attrib["class"],
			'dachs-ex-tapquery literal-block')
	

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