"""
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-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
import io
import functools
import os
import time

from gavo.helpers import testhelpers
from gavo.helpers import trialhelpers

from gavo import base
from gavo import rsc
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 tap_uploads
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(trialhelpers.FakeRequest):
  def __init__(self, *posargs, **kwargs):
  	super().__init__(*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>")
		f.close()

	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>")
		f.close()


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 TAPSchemaQueryTest(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)
		with open(job.getResult("result")[0]) as f:
			res = f.read()
		self.assertTrue('<RESOURCE type="results">' in res)


class _TAPSchemaContent(testhelpers.TestResource):
	def make(self, deps):
		rd = base.parseFromString(rscdesc.RD, """
			<resource schema="schm" resdir="data">
			<meta name="schema-rank">4000</meta>
			<meta name="utype">ivo://ivoa.net/std/cool-model</meta>
			<meta name="description">We actually don't test for empty
				schema descriptions yet</meta>

			<table id="nontap" onDisk="True">
				<column name="col1" description="A column that should not appear
					in tap_schema because its table isn't ADQL-visible"/>
			</table>

			<table id="boring" adql="True">
				<column name="col1" type="spoly"/>
				<column name="quoted/select" type="integer[20]"
					utype="omg:array"/>
				<viewStatement>
					(it's fake)
				</viewStatement>
			</table>

			<table id="tbl" adql="True" nrows="27">
				<meta name="table-rank">1</meta>
				<meta name="utype">adhoc:an.example.table</meta>
				<meta name="description">An example table</meta>
				<foreignKey source="col1,ivoa" inTable="boring" dest="col1,select"/>
				<column name="col1" ucd="src.sample" utype="adhoc:tbl.col1"
					description="A column with a UCD of src.sample"
					unit="km/s"/>
				<column name="ivoa">
					<property name="std">1</property>
				</column>
			</table>
			</resource>""")
	
		rows = {}
		tapRD = base.caches.getRD("//tap")

		# tables is a dispatching grammar
		dd = tapRD.getById("importTablesFromRD")
		for dest, row in dd.grammar.parse(rd):
			del row["parser_"]
			rows.setdefault(dest, []).append(row)

		for ddId in ["importDMsFromRD", "importColumnsFromRD",
				"importFkeysFromRD", "importGroupsFromRD"]:
			dd = tapRD.getById(ddId)
			rows[dd.makes[0].table.id
				] = list(dd.grammar.parse(rd))
		
		return rows


class TAPSchemaCreationTest(testhelpers.VerboseTest):
	resources = [("rows", _TAPSchemaContent())]

	def testSchemaRow(self):
		row = testhelpers.pickSingle(self.rows["schemas"])
		self.assertEqual(row, {
			'schema_description': "We actually don't test for"
				" empty schema descriptions yet",
			'schema_index': '4000',
			'schema_name': 'schm',
			'schema_utype': 'ivo://ivoa.net/std/cool-model'})

	def testRankedTable(self):
		rows = [r for r in self.rows["tables"] if r["table_name"]=="schm.tbl"]
		self.assertEqual(testhelpers.pickSingle(rows), {'nrows': 27,
			'schema_name': 'schm',
			'sourceRD': 'temporary',
			'table_description': 'An example table',
			'table_index': '1',
			'table_name': 'schm.tbl',
			'table_type': 'table',
			'table_utype': 'adhoc:an.example.table'})

	def testNonADQLTable(self):
		for r in self.rows["tables"]:
			if "nontap" in r["table_name"]:
				raise AssertionError("non-ADQL table in tap_schema?")
	
	def testPlainTable(self):
		row = testhelpers.pickSingle(
			[r for r in self.rows["tables"] if r["table_name"]=="schm.boring"])
		self.assertEqual(row, {'nrows': None,
			'schema_name': 'schm',
			'sourceRD': 'temporary',
			'table_description': "We actually don't test for empty schema"
				" descriptions yet",
			'table_index': None,
			'table_name': 'schm.boring',
			'table_type': 'view',
			'table_utype': None})

	def testColumnUtype(self):
		row = testhelpers.pickSingle([r for r in self.rows["columns"]
			if r["utype"]=="adhoc:tbl.col1"])
		self.assertEqual(row["datatype"], "float")
		self.assertEqual(row["column_name"], "col1")
		self.assertEqual(row["ucd"], "src.sample")
		self.assertEqual(row["utype"], "adhoc:tbl.col1")
		self.assertEqual(row["unit"], "km/s")
		self.assertEqual(row["description"],
			"A column with a UCD of src.sample")

	def testXtype(self):
		row = testhelpers.pickSingle([r for r in self.rows["columns"]
			if r["xtype"]=="polygon"])
		self.assertEqual(row["column_name"], "col1")
		self.assertEqual(row["xtype"], "polygon")
		self.assertEqual(row["datatype"], "double")
		self.assertEqual(row["arraysize"], "*")
	
	def testQuotedColname(self):
		row = testhelpers.pickSingle([r for r in self.rows["columns"]
			if r["utype"]=="omg:array"])
		self.assertEqual(str(row["column_name"]), '"select"')
	
	def testSizedArray(self):
		row = testhelpers.pickSingle([r for r in self.rows["columns"]
			if r["utype"]=="omg:array"])
		self.assertEqual(row["arraysize"], '20')
		self.assertEqual(row["datatype"], "int")
		self.assertEqual(row["size"], 20)

	def testStd(self):
		row = testhelpers.pickSingle([r for r in self.rows["columns"]
			if r["column_name"]=="ivoa"])
		self.assertEqual(row["std"], 1)

	def testKeys(self):
		# Update the grammar to be dispatching and then this test, too
		rows = self.rows["keys"]
		self.assertEqual(len(rows), 3)
		self.assertEqual(rows[0]["from_table"], "schm.tbl")
		self.assertEqual(rows[0]["target_table"], "schm.boring")

		self.assertEqual(rows[1]["from_column"], "col1")
		self.assertEqual(rows[1]["target_column"], "col1")

		self.assertEqual(rows[2]["from_column"], "ivoa")
		# TODO: select should probably be quoted here.
		self.assertEqual(rows[2]["target_column"], "select")

		self.assertEqual(len(set(r["key_id"] for r in rows)), 1)

# todo?: groups (but then it's been ages since we've last heard of it)

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)
				with open(job.getResult("result")[0]) as f:
					return f.read()
			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("<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(200):
				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 _RoundtrippedData(testhelpers.TestResource):
	resources = [("conn", tresc.dbConnection)]
	
	def make(self, deps):
		conn = deps["conn"]
		data = rsc.makeData(
			base.resolveCrossId("data/test#import_for_roundtrip"),
			connection=conn)
		try:
			return _getQueryResult("SELECT * FROM test.typestable")[0]
		finally:
			data.drop(data.dd, connection=conn)


class TAPRoundTripTest(testhelpers.VerboseTest):
	resources = [("data", _RoundtrippedData())]

	def testIntegers(self):
		self.assertEqual(self.data[0][0], None)
		self.assertEqual(self.data[1][0], 121)

	def testReals(self):
		self.assertEqual(self.data[0][1], None)
		# todo: can we do better and keep this at an actual float32?
		self.assertAlmostEqual(self.data[1][1], 121.12100219726562)
		self.assertEqual(self.data[0][2], None)
		self.assertAlmostEqual(self.data[1][2], 121.121)

	def testText(self):
		# VOTable doesn't know None from ''
		self.assertEqual(self.data[0][3], "")
		self.assertEqual(self.data[1][3], "hundertEinundzwanzig")

	def testDate(self):
		self.assertEqual(self.data[0][4], None)
		self.assertEqual(self.data[1][4], datetime.datetime(1969, 12, 24, 0, 0))

	def testJSON(self):
		# for now, our VOTable parser doesn't know the json xtype; we
		# way want to change this, in which case this test would need to
		# change.
		# Oh: and VOTable doesn't know None from ''; if we do treat the xtype
		# one day, the "" in the next line ought to become a None.
		self.assertEqual(self.data[0][5], "")
		self.assertEqual(self.data[1][5], '{"testing": true}')
		

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

	def make(self, deps):
		return testhelpers.getXMLTree(
			_getUnparsedQueryResult(
				"SELECT rv, PI(), alpha, delta 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, alpha, delta FROM test.adql LIMIT 20000")

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

	def testParamCopied(self):
		pNode = self.tree.xpath("//PARAM[@name='fib']")[0]
		self.assertEqual(pNode.get("utype"), "math:series.fibonacci")
		self.assertEqual(pNode.get("value"), "1 1 2")

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

	def testCitationInfo(self):
		info = self.tree.xpath("//INFO[@name='citation']")[0]
		self.assertEqual(info.get("value"),
			"http://localhost:8080/tableinfo/test.adql#ti-citing")
		self.assertTrue("Advice " in info.text)

	def testFullCoosys(self):
		coosysEl = self.tree.xpath("//COOSYS")[0]
		self.assertEqual(coosysEl.get("epoch"), "J2015.0")
		self.assertEqual(coosysEl.get("system"), "ICRS")
		self.assertEqual(coosysEl.get("ID"), "system")
		self.assertEqual(
			self.tree.xpath("//FIELD[@name='alpha']")[0].get("ref"),
			"system")
		self.assertEqual(
			self.tree.xpath("//FIELD[@name='rV']")[0].get("ref"),
			"system")


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)

TAP_JOIN_RESULT_TABLE = _TAPJoinResultTable()


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

	def testDatalinkServiceMentioned(self):
		svcDefinition = self.tree.xpath("//RESOURCE[@type='meta']")
		idid = testhelpers.pickSingle(svcDefinition
			).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)

	def testParamCopied(self):
		pNode = self.tree.xpath("//PARAM[@name='fib']")[0]
		self.assertEqual(pNode.get("utype"), "math:series.fibonacci")
		self.assertEqual(pNode.get("value"), "1 1 2")

		pNode = self.tree.xpath("//PARAM[@name='fab']")[0]
		self.assertEqual(pNode.get("value"), "four")


class DataOriginTest(testhelpers.VerboseTest):
	resources = [("tree", TAP_JOIN_RESULT_TABLE)]

	def _getINFO(self, key):
		matches = self.tree.xpath(f"//INFO[@name='{key}']")
		assert len(matches)==1
		return matches[0].get("value")

	def _getINFOs(self, key):
		return set(m.get("value")
			for m in self.tree.xpath(f"//INFO[@name='{key}']"))

	# Keep the sequence of tests in sync with the sequence of items in
	# the data origin paper so it's easy to see whether we're covering
	# them all.

	def testAllIVOIDsPresent(self):
		self.assertEqual(self._getINFOs("ivoid"), set([
				'ivo://x-testing/tap',
				'ivo://x-testing/data/ssatest/c',
				'ivo://x-testing/data/ssatest/s',
				"ivo://x-testing/data/ssatest/hcdtest"]))

	def testPublisher(self):
		self.assertEqual(
			self._getINFO("publisher"),
			"Your organisation's name")

	def testServerSoftware(self):
		self.assertEqual(self._getINFO("server_software"),
			base.SERVER_SOFTWARE)

	# We don't do request and request_post in TAP for now; the core
	# doesn't have a request object, and the renderer sees pre-rendered
	# data and can't put it in any more.

	def testRequestDate(self):
		self.assertTrue(self._getINFO("request_date").startswith("20"))

	def testContact(self):
		self.assertEqual(
			self._getINFOs("contact"),
			set(['invalid@wherever.else', 'hendrix@nirvana.int']))

	def testReferenceURL(self):
		self.assertEqual(
			self._getINFOs("reference_url"),
			set([
				'http://localhost:8080/tableinfo/test.adql',
				'http://localhost:8080/tableinfo/test.hcdtest',
				'http://localhost:8080/__system__/tap/run/info']))

	def testAllSourcesPresent(self):
		self.assertEqual(
			self._getINFOs("publication_id"),
			set(["1635QB41.G135......", "2015ivoa.spec.0617D"]))

	def testResourceVersion(self):
		self.assertEqual(
			self._getINFOs("resource_version"),
			set(["3.2.1-testing"]))

	def testRights(self):
		self.assertEqual(
			self._getINFOs("rights"),
			set(['Everything in here is pure fantasy (distributed under the GNU GPL)']))

	def testRightsURI(self):
		self.assertEqual(
			self._getINFOs("rights_uri"),
			set(["http://too-lazy-to/loop/up/spdx"]))

	def testCreator(self):
		self.assertEqual(
			self._getINFOs("creator"),
			set([
				'John C. Testwriter',
				'The Master Tester',
				'Could be same as contact.name',
				'Hendrix, J.; Page, J; et al.']))

	# the rest is non-data origin so far

	def testSQLQuery(self):
		self.assertEqual(
			self._getINFO("sql_query"),
			"SELECT delta, ssa_pubdid FROM test.adql JOIN test.hcdtest ON ( alpha = ssa_length ) LIMIT 20000")
	
	def testADQLQuery(self):
		self.assertEqual(
			self._getINFO("query"),
			"SELECT delta, ssa_pubdid from test.adql join test.hcdtest on (alpha=ssa_length)")

	def testAllCitationsPresent(self):
		self.assertEqual(
			self._getINFOs("citation"),
			set(['http://localhost:8080/tableinfo/test.adql#ti-citing',
				'http://localhost:8080/tableinfo/test.hcdtest#ti-citing',
				'http://localhost:8080/__system__/tap/run/howtocite']))


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 data centre.")
	
	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"], 100)

	def testTablesPublished(self):
		res = list(self.conn.queryToDicts(
			"select * from tap_schema.tables where table_name='test.csdata'"))
		row = testhelpers.pickSingle(res)
		self.assertEqual(row["table_type"], "table")
		# I suppose we ought to create some way (properties?) to set table_index
		self.assertEqual(row["table_index"], None)
		self.assertEqual(row["nrows"], 10)

	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')


class _WeirdTypesTAPResult(testhelpers.TestResource):
	resources = [("table", tresc.weirdTypesTable)]

	def make(self, ignored):
		data, metadata = _getQueryResult(
			"select textarr from test.weirdtypes")
		return list(metadata.iterDicts(data)), metadata


class WeirdTypesTAPTest(testhelpers.VerboseTest):
	resources = [("res", _WeirdTypesTAPResult())]

	def testStringArrayValue(self):
		self.assertEqual(self.res[0][0]['textarr'],
			['ein       ', 'walde     ', 'viel zu la'])

	def testStringArrayMeta(self):
		self.assertEqual(
			self.res[1].votTable.getChildDict()["FIELD"][0].arraysize,
			"10x*")


class UserUploadCleanupTest(testhelpers.VerboseTest):
	def testCleanup(self):
		with base.getWritableAdminConn() as conn:
			with open(base.getConfig("inputsDir") / "data" / "vizier_votable.vot",
					"rb") as f:
				tap_uploads.uploadUserTable(
					conn, "glue", f, "testuser")
				f.seek(0)
				tap_uploads.uploadUserTable(
					conn, "flu", f, "testuser")
			conn.execute("UPDATE tap_user.tables SET expiry=CURRENT_TIMESTAMP"
				" WHERE user_name='testuser' AND table_name='flu'")

		try:
			cleanup = base.resolveCrossId("//tap_user#expire")
			cleanup.runImmediate()
			with base.getTableConn() as conn:
				still_there = [r[0] for r in
					conn.query("select table_name from tap_user.tables"
						" where user_name='testuser'")]
				self.assertEqual(still_there, ["glue"])

				q = base.UnmanagedQuerier(conn)
				self.assertEqual(
					q.getTableType("tap_user._testuser_glue"), "BASE TABLE")
				self.assertEqual(q.getTableType("tap_user._testuser_flu"), None)
		finally:
			with base.getWritableAdminConn() as conn:
				conn.execute("DROP TABLE IF EXISTS tap_user._testuser_glue")
				conn.execute("DROP TABLE IF EXISTS tap_user._testuser_flu")
				conn.execute("DELETE FROM tap_user.tables WHERE user_name='testuser'")


class _UploadedTables(testhelpers.TestResource):
	resources = [("conn", tresc.dbConnection)]

	def make(self, deps):
		conn = deps["conn"]
		with open(base.getConfig("inputsDir") / "data" / "vizier_votable.vot",
				"rb") as f:
			tap_uploads.uploadUserTable(
				conn, "glue", f, "anonymous")
		return conn

	def clean(self, conn):
		createdTables = ['_anonymous_glue']
		conn.execute("DELETE FROM tap_user.tables WHERE"
			" physical_name in ({})".format(",".join(repr(t) for t in createdTables)))
		for t in createdTables:
			conn.execute(f"DROP TABLE tap_user.{t}")


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

	def testInfoGetterSuccess(self):
		getter = adqlglue.DaCHSFieldInfoGetter(forUser="anonymous")
		infos = getter.getInfosFor("tap_user.glue")
		col = infos[1]
		self.assertEqual(col[0].name, "_RAJ2000")
		self.assertEqual(col[1].ucd, "pos.eq.ra;meta.main")

	def testInfoGetterWrongUser(self):
		getter = adqlglue.DaCHSFieldInfoGetter(forUser="testing")
		self.assertRaisesWithMsg(utils.NotFoundError,
			"table 'tap_user.glue' could not be located in your uploaded tables",
			getter.getInfosFor,
			("tap_user.glue",))

	def testADQLMorphing(self):
		res = list(
			adqlglue.runTAPQuery("WITH a AS"
				" (SELECT * from tap_user.glue where mag<1)"
				" SELECT b.tyc2 from tap_user.glue as b join a using (iras)",
			2, self.conn, [], 20, False))
		self.assertEqual(res, [{'TYC2': '1266-01416-1'}])


class _UploadedTablesUsers(testhelpers.TestResource):
	resources = [("conn", tresc.dbConnection)]

	def make(self, deps):
		conn = deps["conn"]
		with open(base.getConfig("inputsDir") / "data" / "vizier_votable.vot",
				"rb") as f:
			tap_uploads.uploadUserTable(
				conn, "glue", f, "anonymous")
	
		with open(base.getConfig("inputsDir") / "data" / "vizier_votable.vot",
				"rb") as f:
			tap_uploads.uploadUserTable(
				conn, "honk", f, "testing")

		with open(base.getConfig("inputsDir") / "data" / "vizier_votable.vot",
				"rb") as f:
			tap_uploads.uploadUserTable(
				conn, "glue", f, "george")

		with open(base.getConfig("inputsDir") / "data" / "vizier_votable.vot",
				"rb") as f:
			tap_uploads.uploadUserTable(
				conn, "flu", f, "testing")

		return conn

	def clean(self, conn):
		createdTables = [
			'_anonymous_glue', "_testing_honk", "_george_glue", "_testing_flu"]
		conn.execute("DELETE FROM tap_user.tables WHERE"
			" physical_name in ({})".format(",".join(repr(t) for t in createdTables)))
		for t in createdTables:
			conn.execute(f"DROP TABLE tap_user.{t}")


TAP_SERVICE = base.resolveCrossId("//tap#run")

class TableListingTest(testhelpers.VerboseTest):
	resources = [("tables", _UploadedTablesUsers())]
	
	def getForUser(self, username="testting", password="testing"):
		req = trialhelpers.FakeRequest("")
		req.user, req.password = username, password
		tap_uploads.PersistentUploadRenderer(req, TAP_SERVICE
			).render(req)
		return testhelpers.getXMLTree(req.accumulator, debug=False)

	def testTablesListingAnonymous(self):
		self.assertEqual(
			self.getForUser("").xpath("//INFO[@name='QUERY_STATUS']")[0].text,
			"No persistent tables listing for anonymous users.")

	def testTablesListingAuthUser(self):
		tree = self.getForUser("testing")
		self.assertEqual(tree.xpath("schema/name")[0].text,  "tap_user")
		tables = tree.xpath("schema/table")
		# this asserts there's no tables from george or from anonymous
		self.assertEqual(len(tables), 2)

		table_names = {t.xpath("name")[0].text for t in tables}
		self.assertEqual(table_names, {"tap_user.honk", "tap_user.flu"})

		ucds = set(n.text for n in tree.xpath("//ucd"))
		self.assertEqual(ucds, {'pos.eq.ra;meta.main', 'meta.id;meta.main',
			'pos.eq.dec;meta.main', 'pos.angDistance', 'phot.color',
			'pos.pm;pos.eq.ra', 'pos.eq.dec', 'meta.id;src', 'pos.eq.ra',
			'phot.mag;em.opt.V', 'pos.pm;pos.eq.dec'})


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