Source code for gavo.helpers.trialhelpers

"""
Helpers for trial-based tests, in particular retrieving pages.
"""

#c Copyright 2008-2023, 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 functools
import os

from twisted.python import failure  #noflake: exported name

from gavo.helpers.testhelpers import ( #noflake: exported names
	getTestRD, DataServer, getXMLTree, tempConfig, testFile)

from gavo import base
base.setConfig("web", "enabletests", "True")

from gavo import rsc
from gavo.formal import testing
from gavo import svcs
from gavo import web
from gavo.helpers import testtricks
from gavo.web import common as webcommon
from gavo.web import root
from gavo.formal.testing import ( #noflake: exported names
	RenderTest, assertHasStrings)


class ArchiveTest(RenderTest):
	"""the main class for writing trial-based tests in DaCHS.

	It comes preconfigured to handle URLs as the running server and will
	return roughly the same error responses.

	It also has a method to XSD-validate documents against DaCHS' built-in
	schemas.
	"""
	renderer = root.ArchiveService()
	errorHandler = staticmethod(webcommon.produceErrorDocument)

	def assertResponseIsValid(self, res):
		errs = testtricks.getXSDErrors(res[0], True)
		if errs:
			with open("remote.data", "wb") as f:
				f.write(res[0])
			raise AssertionError(errs)


@functools.lru_cache(1)
def getImportConnection():
	# we cannot use the connection pools here since they may create threads.
	return base.getDBConnection("admin")


def provideRDData(rdName, ddId, _imported=set()):
	"""makes ddId from rdName and returns a cleanup function.

	This is for creating temporary data for tests; it's supposed to be used
	as in::

		atexit.register(provideRDData("test", "import_fitsprod"))
	
	This keeps track of (rdName, ddId) pairs it's already loaded and
	doesn't import them again.
	"""
	if (rdName, ddId) in _imported:
		return lambda: None

	dd = getTestRD(rdName).getById(ddId)
	conn = getImportConnection()
	dataCreated = rsc.makeData(dd, connection=conn)
	conn.commit()
	_imported.add((rdName, ddId))

	# rsc may already be gone in atexit
	nvArg = rsc.parseNonValidating

	def cleanup():
		dataCreated.dropTables(nvArg)
	
	return cleanup


def makeRequestArgs(aDict=None, **kwargs):
	"""returns a request.args compatible dict form aDict.

	In particular, simple values will be put into lists.
	"""
	if aDict is None:
		aDict = {}
	aDict.update(kwargs)
	return dict((k,v if isinstance(v, list) else [v])
		for k, v in aDict.items())


class FakeRequest(testing.FakeRequest, web.Request):
	# our request class has a few frills that we want to exercise, too,
	# so we mix it into the standard fake request
	pass
testing.FakeRequest = FakeRequest

# for convenience, expose FakeFile, too
FakeFile = testing.FakeFile

def runSvcWith(service, renderer, args):
	"""runs svc through renderers, passing a dict args.

	args maps to single values, *not* to lists as in t.w.  Unless, of course,
	you actually have lists to begin with.
	"""
	args = makeRequestArgs(args)
	queryMeta = svcs.QueryMeta.fromRequestArgs(args)
	return service.run(renderer, args, queryMeta)


runQuery = functools.partial(testing.runQuery,
	produceErrorDocument=webcommon.produceErrorDocument)


if os.environ.get("GAVO_LOG")!="no":
	from gavo.user import logui
	logui.LoggingUI(base.ui)