Source code for gavo.helpers.trialhelpers

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

#c Copyright 2008-2023, the GAVO project <>
#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)

[docs]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)
[docs] def assertResponseIsValid(self, res): errs = testtricks.getXSDErrors(res[0], True) if errs: with open("", "wb") as f: f.write(res[0]) raise AssertionError(errs)
[docs]@functools.lru_cache(1) def getImportConnection(): # we cannot use the connection pools here since they may create threads. return base.getDBConnection("admin")
[docs]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
[docs]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())
[docs]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
[docs]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, 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)