"""
Tests for resource descriptor handling
"""

#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 pathlib
import os
import threading
import time
import unittest

from gavo.helpers import testhelpers

from gavo import base
from gavo import rsc
from gavo import rscdef
from gavo import rscdesc
from gavo import utils
from gavo.base import meta
from gavo.protocols import tap
from gavo.rscdef import regtest

import tresc


class CanonicalizeTest(
		testhelpers.VerboseTest, metaclass=testhelpers.SamplesBasedAutoTest):
# tests for mapping paths and stuff to canonical ids.
	inp = base.getConfig("inputsDir")

	def _runTest(self, sample):
		src, expected = sample
		self.assertEqual(
			rscdesc.canonicalizeRDId(src),
			expected)
	
	samples = [
		("/somewhere/bad", "/somewhere/bad"),
		("/somewhere/bad.crazy", "/somewhere/bad.crazy"),
		("/somewhere/bad.rd", "/somewhere/bad"),
		("//tap", "__system__/tap"),
		("//tap.rd", "__system__/tap"),
#5
		(inp / "where", "where"),
		(inp / "where/q", "where/q"),
		(inp / "where/q.rd", "where/q"),
		("/resources/inputs/where/q.rd", "where/q"),
		("/resources/inputs/where/q", "where/q"),]
	

class InputStreamTest(testhelpers.VerboseTest):
# test the location of input streams.  This assumes testhelpers has set
# gavo_inputs to <test_dir>/data

	def _assertSourceName(self, rdId, expectedSuffix):
		fName, fobj = rscdesc.getRDInputStream(rscdesc.canonicalizeRDId(rdId))
		self.assertTrue(str(fName).endswith(expectedSuffix),
			"%r does not end with %r"%(fName, expectedSuffix))
		fobj.close()

	def testInternalResource(self):
		self._assertSourceName("//users", "/resources/inputs/__system__/users.rd")

	def testUserResource(self):
		self._assertSourceName("data/test", "data/test.rd")
	
	def testUserOverriding(self):
		dirName = base.getConfig("inputsDir") / "__system__"
		try:
			os.mkdir(dirName)
		except os.error: # don't worry if someone left the dir
			pass

		try:
			testName = dirName / "users"
			open(testName, "w").close()
			try:
				self._assertSourceName("//users", str(testName))
			finally:
				os.unlink(testName)
		finally:
			os.rmdir(dirName)


class MetaTest(testhelpers.VerboseTest):
	"""Test for correct interpretation of meta information.
	"""

	@staticmethod
	def _makeRD(moreStuff):
		return base.parseFromString(rscdesc.RD,
			f"""<resource schema="test">
				{tresc.MINIMAL_META}
				{moreStuff}
				</resource>""")

	def testMetaAttachment(self):
		"""tests for proper propagation of meta information.
		"""
		meta.CONFIG_META.addMeta("test.fromConfig", "from Config")

		rd = self._makeRD("""
			<data id="metatest">
				<meta name="onData">present</meta>
				<table id="noname">
					<column name="alpha"
						type="double precision" required="true"/>
					<meta name="test.inRec">from Rec</meta>
				</table>
			</data>""")

		recDef = rd.getTableDefById("noname")
		self.assertTrue(str(recDef.getMeta("test.inRec")), "from Rec")
		self.assertTrue(str(recDef.getMeta("test.inRd")), "from Rd")
		self.assertTrue(str(recDef.getMeta("test.fromConfig")), "from Config")
		self.assertEqual(recDef.getMeta("test.doesNotExist"), None)

	def testComplexMeta(self):
		"""tests for handling of complex meta items.
		"""
		rd = self._makeRD("""<data id="metatest"/>""")

		data = rd.getById("metatest")
		data.addMeta("testStatus", base.META_CLASSES_FOR_KEYS["info"](
			"I'm so well I could cry",
			infoValue="OK"))
		self.assertTrue(
			isinstance(data.getMeta("testStatus").children[0],
			base.MetaInfoItem))
		self.assertEqual(data.getMeta("testStatus").children[0].infoValue, "OK")
		self.assertEqual(str(data.getMeta("testStatus")),
			"I'm so well I could cry")
	
	def testBadMessenger(self):
		rd = self._makeRD("""<meta name="coverage.waveband">pigeons</meta>""")
		self.assertRaisesWithMsg(base.MetaValidationError,
			'Meta structure on RD item (within IO:\'<resource schema="test">'
			' <meta> title:x creationDate:...\', line 1) did not validate:'
			' coverage.waveband meta only admits values from messenger,'
			' but \'pigeons\' is not in there.',
			base.validateStructure,
			(rd,))

	def testGoodMessenger(self):
		base.validateStructure(
			self._makeRD("""<meta name="coverage.waveband">Gamma-ray</meta>"""))

	def testBadContentLevel(self):
		rd = self._makeRD("""<meta name="content.type">artwork</meta>""")
		self.assertRaisesWithMsg(base.MetaValidationError,
			'Meta structure on RD item (within IO:\'<resource schema="test">'
			" <meta> title:x creationDate:...', line 1) did not validate:"
			' content.type meta only admits values from voresource/content_type,'
			" but 'artwork' is not in there.",
			base.validateStructure,
			(rd,))

	def testGoodContentLevel(self):
		base.validateStructure(
			self._makeRD("""<meta name="content.type">Artwork</meta>"""))

	def testBadDateRole(self):
		rd = self._makeRD("""<meta name="_news" role="broken">There goes</meta>""")
		self.assertRaisesWithMsg(base.MetaValidationError,
			'Meta structure on RD item (within IO:\'<resource schema="test">'
			" <meta> title:x creationDate:...', line 1) did not validate:"
			' _news.role meta only admits values from voresource/date_role,'
			" but 'broken' is not in there.",
			base.validateStructure,
			(rd,))

	def testGoodDateRole(self):
		base.validateStructure(
			self._makeRD("""<meta name="_news" role="Available">There goes</meta>"""))




class MacroTest(unittest.TestCase):
	"""Tests for macro evaluation within RDs.
	"""
	def testDefinedMacrosEasy(self):
		rd = base.parseFromString(rscdesc.RD,
			'<resource schema="test"><macDef name="foo">abc</macDef></resource>')
		self.assertEqual(rd.expand("foo is \\foo."), "foo is abc.")

	def testDefinedMacrosWhitespace(self):
		rd = base.parseFromString(rscdesc.RD,
			'<resource schema="test"><macDef name="foo"> a\nbc  </macDef></resource>')
		self.assertEqual(rd.expand("foo is \\foo."), "foo is  a\nbc  .")


class EmbeddedTextTest(testhelpers.VerboseTest):
	def testBadIndent(self):
		rd = base.parseFromString(rscdesc.RD,
			'<resource schema="test">\n'
			'\t<procDef id="goo">\n'
			'\t<code>\n'
			'\t\timport something\n'
			'  break here\n'
			'\t</code>\n'
			'\t</procDef>\n'
			'</resource>')
		self.assertRaisesWithMsg(base.StructureError,
			"Bad indent in line '  break here'",
			rd.getById("goo").getCode,
			())


class TAP_SchemaTest(testhelpers.VerboseTest):
	"""test for working tap_schema export.

	This is another mega test that runs a bunch of functions in sequence.
	We really should have a place to put those.
	"""
	resources = [
		("conn", tresc.dbConnection),
		("adqltable", tresc.adqlTestTable),
		("adqlgeo", tresc.adqlGeoTable)]

	def setUp(self):
		testhelpers.VerboseTest.setUp(self)
		self.rd = testhelpers.getTestRD()

	def tearDown(self):
		tap.unpublishFromTAP(self.rd, self.conn)
		testhelpers.VerboseTest.tearDown(self)

	def _checkPublished(self):
		tables = set(r[0] for r in
			(self.conn.query("select table_name from TAP_SCHEMA.tables where sourcerd"
			" = %(rdid)s", {"rdid": self.rd.sourceId})))
		self.assertEqual(tables, {'test.adqltable', 'test.adqlgeo',
			'test.pgs_siaptable', 'test.csdata', 'test.adql', 'test.weirdtypes',
			'test.typestable'})

		columns = set(r[0] for r in
			(self.conn.query(
				"select column_name from TAP_SCHEMA.columns where sourcerd"
			" = %(rdid)s", {"rdid": self.rd.sourceId})))
		self.assertTrue({'a_point', 'row_id', 'foo', 'a_moc'}.issubset(columns))

		fkeys = set(self.conn.query("select from_table, target_table"
				" from TAP_SCHEMA.keys where sourcerd"
				" = %(rdid)s", {"rdid": self.rd.sourceId}))
		self.assertEqual(fkeys,
			set([('test.adqltable', 'test.adql')]))
		fkcols = set(r for r in
			(self.conn.query("select from_column, target_column"
				" from TAP_SCHEMA.key_columns where sourcerd"
				" = %(rdid)s", {"rdid": self.rd.sourceId})))
		self.assertEqual(fkcols, set([('foo', 'rv')]))

	def _checkUnpublished(self):
		tables = set(r[0] for r in
			(self.conn.query("select table_name from TAP_SCHEMA.tables where sourcerd"
				" = %(rdid)s", {"rdid": self.rd.sourceId})))
		self.assertEqual(tables, set())
		columns = set(r[0] for r in
			(self.conn.query(
				"select column_name from TAP_SCHEMA.columns where sourcerd"
				" = %(rdid)s", {"rdid": self.rd.sourceId})))
		self.assertEqual(columns, set())
		fkeys = set(self.conn.query("select from_table, target_table"
				" from TAP_SCHEMA.keys where sourcerd"
				" = %(rdid)s", {"rdid": self.rd.sourceId}))
		self.assertEqual(fkeys, set())
		fkcols = set(r for r in
			(self.conn.query("select from_column, target_column"
				" from TAP_SCHEMA.key_columns where sourcerd"
				" = %(rdid)s", {"rdid": self.rd.sourceId})))
		self.assertEqual(fkcols, set())

	def testMega(self):
		tap.publishToTAP(self.rd, self.conn)
		self._checkPublished()
		tap.unpublishFromTAP(self.rd, self.conn)
		self._checkUnpublished()


class RestrictionTest(testhelpers.VerboseTest, metaclass=testhelpers.SamplesBasedAutoTest):
	"""Tests for rejection of constructs disallowed in restricted RDs.
	"""

	def _runTest(self, sample):
		context = rscdesc.RDParseContext(restricted=True)
		context.srcPath = os.getcwd()
		self.assertRaises(base.RestrictedElement, base.parseFromString,
			rscdesc.RD, '<resource schema="testing">'
				'<table id="test"><column name="x"/></table>'
				'%s</resource>'%sample,
				context)
		
	samples = [
		'<procDef><code>pass</code></procDef>',
		'<dbCore queriedTable="test"><condDesc><phraseMaker/></condDesc></dbCore>',
		'<nullCore id="null"/><service core="null"><customRF name="foo"/>'
			'</service>',
		'<table id="test2"><column name="x"/><index columns="x">'
			'CREATE</index></table>',
		'<table id="test2"><column name="x"/><index columns="x">'
			'<option>ASC</option></index></table>',
		'<table id="test2"><column name="x" fixup="__+\'x\'"/></table>',
		'<data><embeddedGrammar><iterator/></embeddedGrammar></data>',
		'<data><reGrammar names="a,b" preFilter="rm -rf /"/></data>',
		'<data><directGrammar cBooster="res/kill.c"/></data>',
	]


class _RD1(tresc.FileResource):
	path = "inputs/rd1.rd"
	content = """<resource schema='test'>
			<table id="tmp"><column name="a"/></table>
			<data id="import">
				<dictlistGrammar/>
				<make table="tmp" rowmaker="rd2#rmk"/>
			</data>
		</resource>"""


class _RD2(tresc.FileResource):
	path = "inputs/rd2.rd"
	content = """<resource schema='test'>
			<rowmaker id="rmk" idmaps="*">
				<apply name="foo">
					<code>
						@a = ord(@a)*0.5
					</code>
				</apply>
			</rowmaker>
		</resource>"""


class RestrictionInheritingTest(testhelpers.VerboseTest):
	# tests making sure cross-RD references don't break out of restricted
	# mode
	resources = [("rd1", _RD1()), ("rd2", _RD2())]

	def testRestrictionRequired(self):
		rows = rsc.makeData(base.caches.getRD("rd1"
			).getById("import"),
			forceSource=[{'a': 'x'}]).getPrimaryTable().rows
		self.assertEqual(rows, [{'a': 60.0}])

	def testRestrictionEnforced(self):
		self.assertRaises(base.RestrictedElement,
			base.caches.getRD,
			"rd1", restricted=True)


class _TempRDFile(tresc.FileResource):
	path = "inputs/temp.rd"
	content = "<resource schema='temptemp' resdir='data'/>"


class _UnloadableRDFile(tresc.FileResource):
	# this is handcrafted for specific failure in parallelRDIsInvalidated
	path = "inputs/unloadable.rd"
	content = """<resource schema='temptemp' resdir="data">
			<LOOP>
				<codeItems>
					from gavo import base
					base.OTHER_IS_WAITING = True
					while True:
						if hasattr(base, "I_AM_HERE"):
							raise base.ReportableError("All is good")
					yield {"foo": "a"}
				</codeItems>
				<events>
					<table id="\\foo"/>
				</events>
			</LOOP>
		</resource>"""


class _TempBadRDFile(tresc.FileResource):
	path = "inputs/tempbad.rd"
	content = "<resource schema='temptemp' resdir='data'>junk</resource>"


class CachesTest(testhelpers.VerboseTest):
	resources = [("tempRDFile", _TempRDFile()),
		("tempBadRDFile", _TempBadRDFile())]

	def testCacheWorks(self):
		rd1 = base.caches.getRD("//users")
		rd2 = base.caches.getRD("//users")
		self.assertTrue(rd1 is rd2)

	def testNormalisationRD(self):
		rd1 = base.caches.getRD("temp")
		other = base.caches.getRD("temp.rd")
		self.assertTrue(rd1 is other)

	def testNormalisationStripInputs(self):
		rd1 = base.caches.getRD("temp")
		other = base.caches.getRD(
			os.path.join(base.getConfig("inputsDir"), "temp.rd"))
		self.assertTrue(rd1 is other)
	
	def testNormalisationImmediatePath(self):
		rd1 = base.caches.getRD("temp")
		other = base.caches.getRD(
			os.path.join(base.getConfig("inputsDir"), ".", ".", "temp.rd"))
		self.assertTrue(rd1 is other)

	def testNormalisationParentDir(self):
		rd1 = base.caches.getRD("temp")
		other = base.caches.getRD(
			os.path.join(base.getConfig("inputsDir"), "..", "inputs", "temp.rd"))
		self.assertTrue(rd1 is other)

	def testNormalisationFailure(self):
		self.assertRaisesWithMsg(ValueError,
			"Too many .. in relative rd id",
			base.caches.getRD,
			("../inputs/temp",))

	def testCachesCleared(self):
		rd1 = base.caches.getRD("//users")
		rd1.getById("users").gobble = "funk"
		base.caches.clearForName(rd1.sourceId)
		rd2 = base.caches.getRD("//users")
		self.assertFalse(rd2 is rd1)
		self.assertTrue(hasattr(rd1.getById("users"), "gobble"))
		self.assertFalse(hasattr(rd2.getById("users"), "gobble"))

	def testAliases(self):
		rd1 = base.caches.getRD("//users")
		rd1.getById("users").gobble = "funk"
		base.caches.clearForName("__system__/users")
		rd2 = base.caches.getRD("//users")
		rd3 = base.caches.getRD("__system__/users.rd")
		self.assertFalse(rd2 is rd1)
		self.assertFalse(rd1 is rd3)
		self.assertTrue(hasattr(rd1.getById("users"), "gobble"))
		self.assertFalse(hasattr(rd2.getById("users"), "gobble"))

	def testDirty(self):
		origRD = base.caches.getRD("temp")
		sameRD = base.caches.getRD("temp")
		self.assertTrue(origRD is sameRD)
		now = time.time()
		origRD.loadedAt = now-30
		os.utime(self.tempRDFile.original, (now+1, now+1))
		otherRD = base.caches.getRD("temp")
		self.assertFalse(origRD is otherRD)

	def testExceptionsAreCached(self):
		try:
			base.caches.getRD("tempbad")
		except base.StructureError as ex:
			ex1 = ex
		try:
			base.caches.getRD("tempbad")
		except base.StructureError as ex:
			ex2 = ex
		self.assertTrue(ex1 is ex2)

	def testReloadAttemptForBad(self):
		base.caches.clearForName("tempbad")
		try:
			base.caches.getRD("tempbad")
		except base.StructureError as ex:
			ex1 = ex
		os.utime(os.path.join(base.getConfig("inputsDir"), "tempbad.rd"),
			(time.time()+1, time.time()+1))
		try:
			_ = base.caches.getRD("tempbad")
		except base.StructureError as ex:
			self.assertTrue(ex is ex1)
		else:
			self.fail("This should have bombed out?")

	def testTemporarilyBadRDsRemainCached(self):
		base.caches.clearForName("flapping")

		with testhelpers.testFile("flapping.rd",
				"<resource schema='murk' resdir='data'/>",
				inDir=base.getConfig("inputsDir")):
			self.assertEqual(base.caches.getRD("flapping").schema, "murk")

		base.caches.getRD("flapping").loadedAt = time.time()-30

		with testhelpers.testFile("flapping.rd",
				"broken",
				inDir=base.getConfig("inputsDir"),
				timestamp=int(time.time()+1)):
			self.assertEqual(base.caches.getRD("flapping").schema, "murk")

		base.caches.getRD("flapping").loadedAt = time.time()-30

		with testhelpers.testFile("flapping.rd",
				"<resource schema='gurk' resdir='data'/>",
				inDir=base.getConfig("inputsDir"),
				timestamp=int(time.time()+2)):
			self.assertEqual(base.caches.getRD("flapping").schema, "gurk")

	def testRemovedRDRemainsCached(self):
		with testhelpers.testFile("fugit.rd",
				"<resource schema='look' resdir='data'/>",
				inDir=base.getConfig("inputsDir")) as rdName:
			rd = base.caches.getRD(rdName)
			self.assertEqual(rd.schema, "look")
		self.assertEqual(base.caches.getRD(rdName).schema, "look")
		self.assertRaises(base.RDNotFound,
			rscdesc.getRD,
			rdName)

	def testAddedRDIsPickedUp(self):
		with testhelpers.fakeTime(time.time()-60):
			try:
				rd = base.caches.getRD("temporary")
			except base.NotFoundError:
				pass
			else:
				raise AssertionError("Stale temporary.rd in my inputs")

		from gavo import svcs
		clearer = base.resolveCrossId("//services#cacheclear").core
		clearer.run(
			None,
			svcs.CoreArgs(clearer.inputTable, {"rdId": "temporary"}, None),
			None)

		with testhelpers.testFile("temporary.rd",
				"<resource schema='look' resdir='data'/>",
				inDir=base.getConfig("inputsDir")):
			rd = base.caches.getRD("temporary")
			self.assertEqual(rd.schema, "look")


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

	def testRecreateAfter(self):
		q = base.UnmanagedQuerier(connection=self.conn)
		rd = testhelpers.getTestRD()
		t0 = rsc.TableForDef(rd.getById("pythonscript"), connection=self.conn,
			create=False)
		t0.drop()
		self.assertTrue(q.getTableType("test.pythonscript") is None)
		_ = rsc.makeDependentsFor([rd.getById("recaftertest")],
			rsc.parseNonValidating, self.conn, True)
		self.assertFalse(q.getTableType("test.pythonscript") is None)
		self.conn.rollback()

	def testFailedDependencyNonFatal(self):
		dd = base.parseFromString(rscdef.DataDescriptor,
			'<data recreateAfter="data/test#foobarmatic"/>')

		with testhelpers.collectedEvents("Warning") as msgs:
			rsc.makeDependentsFor([dd],
				rsc.parseNonValidating, self.conn, True)
		# obscore may or not have stats at this point, and it will
		# produce a warning if it has not; take that warning out
		msgs = [m for m in msgs if not "Cannot publish product types" in m]
		self.assertEqual(
			testhelpers.pickSingle(msgs)[1],
				"Ignoring dependent data/test#foobarmatic"
				" of None (Element with id 'foobarmatic' could not be located"
				" in RD data/test)")

	def testRecursiveDepedency(self):
		rd = base.parseFromString(rscdesc.RD, """
			<resource schema="test">
			<table id="made"/>
			<rowmaker id="add_fu">
				<apply>
					<code>
						targetTable.tableDef.rd.dataMade.append(vars["dn"])
					</code>
				</apply>
			</rowmaker>
			<data id="stuff0" recreateAfter="stuff1">
				<sources items="1"/>
				<embeddedGrammar id="g">
					<iterator>
						<code>
							yield {"dn": self.grammar.parent.id}
						</code>
					</iterator>
				</embeddedGrammar>
				<make table="made" rowmaker="add_fu"/>
			</data>
			<data original="stuff0" id="stuff1">
				<recreateAfter>stuff2</recreateAfter>
				<recreateAfter>stuff3</recreateAfter>
			</data>
			<data id="stuff2" original="stuff0"/>
			<data id="stuff3" original="stuff0"/>
			</resource>
			""")
		rd.dataMade = []
		_ = rsc.makeDependentsFor([rd.getById("stuff0")],
			rsc.parseNonValidating, self.conn, True)
		self.assertEqual(set(rd.dataMade), set(["stuff1", "stuff2", "stuff3"]))
	
	def testCyclicDependency(self):
		rd = base.parseFromString(rscdesc.RD, """
			<resource schema="test">
			<table id="made"/>
			<rowmaker id="add_fu">
				<apply>
					<code>
						targetTable.tableDef.rd.dataMade.append(vars["dn"])
					</code>
				</apply>
			</rowmaker>
			<data id="stuff0" recreateAfter="stuff1">
				<sources items="1"/>
				<embeddedGrammar id="g">
					<iterator>
						<code>
							yield {"dn": self.grammar.parent.id}
						</code>
					</iterator>
				</embeddedGrammar>
				<make table="made" rowmaker="add_fu"/>
			</data>
			<data original="stuff0" id="stuff1" dependents="stuff2"/>
			<data original="stuff0" id="stuff2" recreateAfter="stuff1"/>
			</resource>
			""")
		rd.dataMade = []
		self.assertRaisesWithMsg(base.ReportableError,
			"Could not sort dependent DDs topologically (use  --hints to learn more).",
			rsc.makeDependentsFor,
			([rd.getById("stuff0")], rsc.parseNonValidating, self.conn, True))

	def testSequencing(self):
		rd = base.parseFromString(rscdesc.RD, """
			<resource schema="test">
				<table id="made" onDisk="True" temporary="True"/>
				<STREAM id="make">
					<make table="made"><script type="preIndex" lang="python">
						table.tableDef.rd.dataMade.append("\\tag")
					</script></make>
				</STREAM>

				<data id="d1" recreateAfter="d3">
					<recreateAfter>d2</recreateAfter><recreateAfter>d4</recreateAfter>
					<FEED source="make" tag="1"/>
				</data>

				<data id="d2">
					<recreateAfter>d4</recreateAfter><recreateAfter>d3</recreateAfter>
					<FEED source="make" tag="2"/>
				</data>

				<data id="d3">
					<recreateAfter>d4</recreateAfter>
					<FEED source="make" tag="3"/>
				</data>

				<data id="d4">
					<recreateAfter>d5</recreateAfter>
					<FEED source="make" tag="4"/>
				</data>

				<data id="d5">
					<FEED source="make" tag="end"/>
				</data>
			</resource>""")

		rd.dataMade = []
		_ = rsc.makeDependentsFor([rd.getById("d1")],
			rsc.parseNonValidating, self.conn, True)
		self.assertEqual(rd.dataMade, ["2", "3", "4", "end"])

	def testRebuildSkipsDefault(self):
		rd =  base.parseFromString(rscdesc.RD, """
			<resource schema="test">
				<table id="rebuild_skip_test" onDisk="True"/>

				<data id="d1" recreateAfter="d2" updating="True">
					<make table="rebuild_skip_test"/>
				</data>

				<data id="d2">
					<make>
						<table id="foo" onDisk="True"/>
						<script type="postCreation" lang="python">
							table.tableDef.rd.testmsg = "d2 ran"
						</script>
					</make>
				</data>

			</resource>""")
		rd.sourceId = "rbs"

		rsc.TableForDef(rd.getById("rebuild_skip_test"),
			create=False,
			connection=self.conn).drop()

		try:
			rsc.makeData(
				rd.getById("d1"),
				connection=self.conn,
				runCommit=False)
			self.assertEqual(rd.testmsg, "d2 ran")

			rd.testmsg = "all is fine"
			rsc.makeData(
				rd.getById("d1"),
				connection=self.conn,
				runCommit=False)

			self.assertEqual(rd.testmsg, "all is fine")
		finally:
			self.conn.rollback()

	def testRebuildWithRemakeOnData(self):
		rd =  base.parseFromString(rscdesc.RD, """
			<resource schema="test">
				<table id="rebuild_skip_test" onDisk="True"/>

				<data id="d1" recreateAfter="d2" updating="True">
					<make table="rebuild_skip_test"/>
				</data>

				<data id="d2" remakeOnDataChange="True">
					<make>
						<table id="foo" onDisk="True"/>
						<script type="postCreation" lang="python">
							table.tableDef.rd.testmsg = "d2 ran"
						</script>
					</make>
				</data>

			</resource>""")
		rd.sourceId = "rbs"

		rsc.TableForDef(rd.getById("rebuild_skip_test"),
			create=False,
			connection=self.conn).drop()

		try:
			rsc.makeData(
				rd.getById("d1"),
				connection=self.conn,
				runCommit=False)
			self.assertEqual(rd.testmsg, "d2 ran")

			rd.testmsg = "all is fine"
			rsc.makeData(
				rd.getById("d1"),
				connection=self.conn,
				runCommit=False)

			self.assertEqual(rd.testmsg, "d2 ran")
		finally:
			self.conn.rollback()


class ConcurrentRDTest(testhelpers.VerboseTest):
	resources = [("unloadableRDFile", _UnloadableRDFile())]

	def testInvalidation(self):
		rd = base.parseFromString(rscdesc.RD, '<resource schema="test"/>')
		rd.sourceId = "artificial"
		rd.invalidate()
		try:
			rd.sourceId
		except base.ReportableError as ex:
			self.assertEqual(str(ex), "Loading of artificial"
				" failed in another thread; this RD cannot be used here")
		else:
			self.fail("Invalidation of an RD didn't work")

	def testRacingClear(self):
		# this is a bit non-deterministic...
		_ = base.caches.getRD("__system__/services")
		base.caches.clearForName("__system__/services")

		chaosInThread = False  #noflake: evaluated elsewhere
		def loadFromOne():
			rd = base.caches.getRD("__system__/services")
			if not hasattr(rd, "serviceIndex"):
				chaosInThread = True  #noflake: evaluated elsewhere

		t1 = threading.Thread(target=loadFromOne)
		t1.daemon = True
		t1.start()
		for retry in range(300):
			if "__system__/services" in rscdesc._CURRENTLY_PARSING:
				break
			time.sleep(0.01)
		else:
			self.fail("getRD in thread does not load the services RD?")
		
		base.caches.clearForName("__system__/services")
		newRD = base.caches.getRD("__system__/services")
		# serviceIndex is created very late in RD creation; if it's missing
		# here, the RLock preventing out-of thread returns of the currently
		# parsing RD is broken.
		self.assertTrue(hasattr(newRD, "serviceIndex"))
		t1.join(10)
		self.assertFalse(chaosInThread, "RD not completed in thread")

	def testParallelRDIsInvalidated(self):
		base.caches.clearForName("unloadable")
		fromThreads = []
		def loadFromOne():
			try:
				fromThreads.append(base.caches.getRD("unloadable"))
			except Exception as ex:
				fromThreads.append(ex)
		t1 = threading.Thread(target=loadFromOne)
		t1.start()

		# Check the _UnloadableRDFile above: This will set base.OTHER_IS_WAITING
		# and the wait for base.I_AM_HERE to appear.  This will happen in
		# thread 1.  We here provide the trigger when we've started to
		# in another thread.
		t2 = threading.Thread(target=loadFromOne)
		t2.start()

		while not hasattr(base, "OTHER_IS_WAITING"):
			time.sleep(0.001)
		base.I_AM_HERE = True
		t1.join(0.2)
		t2.join(0.2)

		# now one of the results must be an exception, the other a
		# BrokenClass
		self.assertEqual(set(r.__class__.__name__ for r in fromThreads),
			frozenset(["BrokenClass", "ReportableError"]))
		

_RUNNERS_RESPONSES = {
	"http://localhost:8080/bar": (200, {},
		b'<VOTABLE version="1.2" xmlns="http://www.ivoa.net/xml/VOTable/v1.2"'
		b' xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"'
		b' xsi:schemaLocation="http://www.ivoa.net/xml/VOTable/v1.2 '
		b'http://vo.ari.uni-heidelberg.de/docs/schemata/VOTable-1.2.xsd">'
		b'<DESCRIPTION>The apparent places</DESCRIPTION><RESOURCE type="meta">'
		b'<DESCRIPTION>give exact</DESCRIPTION></RESOURCE></VOTABLE>'),
	"ivo://ivoa.net/std/quack": (200, {}, b""),
	"http://localhost:8080/data/regtest/foo?testParam=10%25w%2Fo+tax":
		(200, {}, b"Hume and Kant singing in perfect harmony."),
	"http://localhost:8080/data/regtest/nork?urk=zoo&oo=1&oo=2":
		(200, {}, b""),
}


def _fakeRetrieveResource(self, serverURL, timeout=0, moreHeaders=""):
	"""A stand-in for the test runner's testOpener.
	"""
	self.httpURL = self.getValue(base.getConfig("web", "serverURL"))
	try:
		return _RUNNERS_RESPONSES[self.httpURL]
	except KeyError:
		return 404, {}, "Not found"


class _RunnersSample(testhelpers.TestResource):
	def make(self, dependents):
		self.originalRetrieve = regtest.DataURL.retrieveResource
		regtest.DataURL.retrieveResource = _fakeRetrieveResource
		rd = base.caches.getRD("data/regtest")
		return rd

	def clean(self, rsc):
		regtest.DataURL.retrieveResource = self.originalRetrieve


class _RegtestTest(testhelpers.VerboseTest):
	resources = [("rd", _RunnersSample())]

	def assertContains(self, needle, haystack):
		if needle not in haystack:
			raise AssertionError("%s not in %s"%(needle, haystack))


class MiscRegtestTest(_RegtestTest):
	def testBasic(self):
		runner = regtest.TestRunner([self.rd.tests[0]], verbose=False)
		runner.runTestsInOrder()
		self.assertContains("3 of 4 bad.  avg", runner.stats.getReport())

	def testRelativeSource(self):
		self.assertEqual(self.rd.tests[1].tests[0].url.getValue("http://foo"),
			"http://foo/data/regtest/foo?testParam=10%25w%2Fo+tax")

	def testRelativeSourceWithParam(self):
		self.assertEqual(self.rd.tests[1].tests[3].url.getValue("http://foo"),
			"http://foo/data/regtest/nork?urk=zoo&oo=1&oo=2")

	def testAbsoluteSource(self):
		self.assertEqual(self.rd.tests[1].tests[1].url.getValue("http://foo"),
			"http://foo/bar")

	def testURISource(self):
		self.assertEqual(self.rd.tests[1].tests[2].url.getValue("http://foo"),
			"ivo://ivoa.net/std/quack")
		self.assertEqual(self.rd.tests[1].tests[2].url.getParams(),
			[("gobba", "&?")])


class RegtestRunTest(_RegtestTest):
	def testRunWholeRD(self):
		proc, stdout, stderr = testhelpers.captureOutput(regtest.main,
			args=(["-v", "data/regtest"],))
		self.assertContains("**** Test failed: Failing Test -- http://localhost:8",
			stdout)
		self.assertContains("3 of 9 bad.  avg", stdout)
		self.assertEqual(stderr.count("."), 6)
		self.assertEqual(stderr.count("E"), 3)

	def testRunSuite(self):
		proc, stdout, stderr = testhelpers.captureOutput(regtest.main,
			args=(["data/regtest#urltests"],))
		self.assertContains("0 of 5 bad.  avg", stdout)
	
	def testRunSingle(self):
		proc, stdout, stderr = testhelpers.captureOutput(regtest.main,
			args=(["-v", "data/regtest#atest"],))
		self.assertContains("0 of 1 bad.  avg", stdout)

	def testVerboseFailing(self):
		proc, stdout, stderr = testhelpers.captureOutput(regtest.main,
			args=(["-v", "data/regtest#failtest"],))
		self.assertContains("Test failed: Failing Test -- "
			"http://localhost:8080/data/regtest/foo?testParam=10%25w%2Fo+tax",
			stdout)
		self.assertContains(">>>> 'Wittgenstein' missing", stdout)
		self.assertContains("1 of 1 bad.  avg", stdout)

	def testXSDFailing(self):
		proc, stdout, stderr = testhelpers.captureOutput(regtest.main,
			args=(["-v", "data/regtest#xsdfail"],))
		self.assertContains(">>>> Response not XSD valid.  Validator output"
			" starts with\n",
			stdout)

	def testXpathFailing(self):
		proc, stdout, stderr = testhelpers.captureOutput(regtest.main,
			args=(["-v", "data/regtest#xpathfail"],))
		self.assertContains(" not lots -- http://localhost:8080/bar\n\n"
			">>>> Trouble with type: //v2:RESOURCE[1] ('lots', 'meta')\n"
			"1 of 1 bad.",
			stdout)
	
	def testTagSelects(self):
		proc, stdout, stderr = testhelpers.captureOutput(regtest.main,
			args=(["-t", "elite", "data/regtest#dumbsuite"],))
		self.assertContains("4 of 5 bad.", stdout)

	def testWrongTagSelectsNot(self):
		proc, stdout, stderr = testhelpers.captureOutput(regtest.main,
			args=(["-t", "bigserver", "data/regtest#dumbsuite"],))
		self.assertContains("3 of 4 bad.", stdout)


class _CombinedData(testhelpers.TestResource):
	resources = [("csTable", tresc.csTestTable),
		("randomTable", tresc.randomDataTable)]

	def make(self, dependents):
		return rsc.makeCombinedData(testhelpers.getTestRD(
			).getById("import_pythonscript"),
			{"primary": dependents["csTable"], "other": dependents["randomTable"]})


class CombiningDataTest(testhelpers.VerboseTest):
	resources = [("data", _CombinedData())]

	def testTablesIterating(self):
		names = [t.tableDef.id for t in self.data]
		self.assertEqual(set(names), {'randomDataTable', 'csdata'})
	
	def testMetaFromBasis(self):
		self.assertEqual(base.getMetaText(self.data.dd, "onData"), "present")

	def testPrimaryTable(self):
		res = self.data.getPrimaryTable()
		self.assertEqual(list(res), [
			{'tinyflag': None, 'alpha': 10.0, 'rV': None,
			'mag': 14.0, 'delta': 12.0, 'internal': None}])
	
	def testSecondaryTable(self):
		self.assertEqual(self.data.getTableWithRole("other").rows[0],
			{'n': 23, 'x': 29.25})


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

	def make(self, deps):
		rd = base.caches.getRD("data/seeded.rd")
		conn = deps["conn"]

		from gavo.user import importing
		importing.process(rsc.parseNonValidating, ["data/seeded"])

		from gavo.registry import publication
		publication.updateServiceList([rd], connection=conn)

		from gavo.user import limits
		limits.updateForRD(rd, conn)
		
		conn.execute("UPDATE dc.rdmeta SET"
			" spatial='3/5 67 8/2345 9/',"
			" spectral='{1e-19, 2e-19, 5e-20, 6e-20}',"
			" temporal='{20, 30}'"
			" WHERE sourcerd='data/seeded'")
		conn.commit()

		base.caches.clearForName(rd.sourceId)
		return rscdesc.getRD("data/seeded.rd")


class InjectionTest(testhelpers.VerboseTest):
	resources = [("rd", _ImportedAndPublishedRD())]

	def testResMetadataUpdated(self):
		fdate = utils.formatISODT(
			datetime.datetime.utcfromtimestamp(
				os.path.getmtime(
					os.path.join(base.getConfig("inputsDir"), "data/seeded.rd"))))
		resdate = base.getMetaText(self.rd, "_metadataUpdated")
		self.assertEqual(fdate, resdate)

	def testSpatialCoverageInjected(self):
		self.assertEqual(self.rd.coverage.spatial, "3/5 67 8/2345 9/")
		self.assertAlmostEqual(
			self.rd.coverage.parsedMOC.moc.area,
			0.03274090244143415)

	def testInlineCoveragePriority(self):
		# note that there's an UPDATE above that injects something else
		# into the RD.
		self.assertEqual(self.rd.coverage.temporal,
			[(51544.0, 51575.0), (54322.25, 54333.75)])

	def testIntervalsInjected(self):
		self.assertEqual(self.rd.coverage.spectral,
			[(1e-19, 2e-19), (5e-20, 6e-20)])

	def testResDataUpdated(self):
		# the stuff was imported a very short while ago, so we shouldn't be
		# more than a few seconds apart
		impdate = utils.parseISODT(
			base.getMetaText(self.rd, "_dataUpdated"))
		timeSinceImport = (datetime.datetime.utcnow()-impdate).total_seconds()
		self.assertTrue(timeSinceImport<120, "Time between import and test"
			" %s seconds?"%timeSinceImport)

	def testSvcMetadataUpdated(self):
		pubdate = utils.parseISODT(
			base.getMetaText(self.rd.getById("hooray"), "_metadataUpdated"))
		# in contrast to the RD timestamp, that's the time since the last
		# dachs pub, so again it should be rather short
		timeSincePub = (datetime.datetime.utcnow()-pubdate).total_seconds()
		self.assertTrue(timeSincePub<120, "Time between dachs pub and test"
			" %s seconds?"%timeSincePub)

	def testTableMetadataUpdated(self):
		pubdate = utils.parseISODT(
			base.getMetaText(self.rd.getById("seedy"), "_metadataUpdated"))
		# see testSvcMetadataUpdated
		timeSincePub = (datetime.datetime.utcnow()-pubdate).total_seconds()
		self.assertTrue(timeSincePub<120, "Time between dachs pub and test"
			" %s seconds?"%timeSincePub)

	def testTableDataUpdated(self):
		# tables pubs shouldn't have _dataUpdated meta of their own
		impdate = base.getMetaText(self.rd.getById("seedy"), "_dataUpdated")
		self.assertEqual(impdate, base.getMetaText(self.rd, "_dataUpdated"))

	def testNRows(self):
		self.assertEqual(self.rd.getById("seedy").nrows, 10)


class BlacklistTest(testhelpers.VerboseTest):
	def testBasic(self):
		with testhelpers.testFile("q.rd",
				"<resource/>",
				inDir=os.path.join(base.getConfig("inputsDir"), "fies")):
			with testhelpers.tempConfig(
					("general", "rdblacklist", "/fies/q.rd")):
				self.assertRaisesWithMsg(base.Error,
					"RD fies/q is blacklisted in gavo.rc.  Not loading.",
					rscdesc.getRD,
					("fies/q",))

	def testWithExtension(self):
		with testhelpers.testFile("q.rd",
				"<resource/>",
				inDir=os.path.join(base.getConfig("inputsDir"), "fies")):
			with testhelpers.tempConfig(
					("general", "rdblacklist", "/fies/q.rd")):
				self.assertRaisesWithMsg(base.Error,
					"RD fies/q.rd is blacklisted in gavo.rc.  Not loading.",
					rscdesc.getRD,
					("fies/q.rd",))
	
	def testMulti(self):
		with testhelpers.testFile("das.rd",
				"<resource/>",
				inDir=os.path.join(base.getConfig("inputsDir"), "auch")):
			with testhelpers.tempConfig(
					("general", "rdblacklist", "/fies/q.rd, /auch/das.rd")):
				self.assertRaisesWithMsg(base.Error,
					"RD auch/das is blacklisted in gavo.rc.  Not loading.",
					rscdesc.getRD,
					("auch/das",))


class ResdirInferenceTest(testhelpers.VerboseTest):
	def testFromSchema(self):
		ipd = base.getConfig("inputsDir")
		with testhelpers.testFile("bla.rd", '<resource schema="data"/>',
				inDir=ipd / "data/nested/directories"):
			rd = rscdesc.getRD("data/nested/directories/bla.rd")
			self.assertEqual(rd.resdir, ipd / "data")
	
	def testExplicit(self):
		ipd = base.getConfig("inputsDir")
		with testhelpers.testFile("bla.rd",
				'<resource schema="data" resdir="hunks"/>',
				inDir=ipd / "data/nested/directories"):
			rd = rscdesc.getRD("data/nested/directories/bla.rd")
		self.assertEqual(rd.resdir, base.getConfig("inputsDir") / "hunks")

	def testFullPath(self):
		rd = base.parseFromString(rscdesc.RD,
			'<resource schema="foo" resdir="/usr/local/bin"/>')
		self.assertEqual(rd.resdir, pathlib.Path("/usr/local/bin"))

	def testWithDot(self):
		ipd = base.getConfig("inputsDir")
		with testhelpers.testFile("bla.rd",
				'<resource schema="data" resdir="."/>',
				inDir=ipd / "data/nested/directories"):
			rd = rscdesc.getRD("data/nested/directories/bla.rd")
			self.assertEqual(rd.resdir, ipd / "data/nested/directories")


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

	def testTextarr(self):
		recs = list(self.conn.queryToDicts(
			"SELECT * FROM test.weirdtypes WHERE tg='textarr'"))
		self.assertEqual(recs[0]["textarr"],
			["ein", "walde", "viel zu lang..."])


class RunOnTest(testhelpers.VerboseTest):
	def testGoodLoaded(self):
		rd = base.parseFromString(rscdesc.RD, """
			<resource schema="test">
				<execute on="loaded" title="add sentinel"><job><code>
					def getSentinel():
						return "it's there"
					rd.getSentinel = getSentinel
				</code></job></execute>
			</resource>""")
		self.assertEqual(rd.getSentinel(), "it's there")


if __name__=="__main__":
	import warnings
	warnings.filterwarnings("error", category=FutureWarning)
	testhelpers.main(RegtestRunTest)