""" Twisted trial-based tests for TAP and UWS. Tests not requiring the TAP renderer go to taptest.py The real big integration tap tests use the HTTP interface and are in an external resource package taptest. """ #c Copyright 2008-2024, the GAVO project #c #c This program is free software, covered by the GNU GPL. See the #c COPYING file in the source distribution. import atexit import datetime import io import re import os import time from twisted.internet import reactor from twisted.python import threadable threadable.init() from gavo.helpers import trialhelpers from gavo import base from gavo import rscdesc #noflake: for registration from gavo import utils from gavo.protocols import scs #noflake: for table's q3c mixin from gavo.protocols import tap from gavo.protocols import uws base.DEBUG = True _B = "__system__/tap/run/" votWithOldGeom = b'eJxdUctugzAQvOcrVj72YAfUEwIkClRCigIiNNdqC26DRDCxHUj69bVDUKNcPLPa2dmH/X1eRW+b\nFEYuVSv6gDjUJSu4HLteBeSg9eAxNk0TbUeBtOeamRQzVfjVcTZadbjyy3SXf5RxaujNzuB7lm4S\nyJKAKIWfnahRG38CKCVeVfvLA/JCoEGN+jqYoD6gJNDjkT9XnOsmIINQlJ9M0Lc6IA3/MXQuVLr2\nIqWliIWQjaKFUK0tdBO6x+7MXQKXWYnNqfOKPNtWduYk3cVlVlRZvg2zuNzB0hHEN+DApT5LDjXv\nNZc+e1SvfHbbzppEVbQsPXMAvyotWJKEyzRwa+HS9doB59WCz0zaypnVm/ffw7S743JN9nBhdv+z\n8A+9NIXB\n' votWithDALIGeom = b'eJxdkcFugzAMhu88hZUHSFq2UwVIDJiEVI0KWK9TBpmGRBOWBGj39HNK0ape8tvJ599xEhyLOn7Z\nZzAJbTolQ7KlT8SD86mXJiTf1g47xuZ5pt2kOJXCMjxiWMU/e8EmR0deUGZV8V4mGYZXO9TXPNun\nkKchMYZ/9KrhFv0JcK35xXS/IiQ+gZZbbi8DJq0a0ZGA5CfxWDM2bUgGZaj4wWThjW12sbFaJUrp\n1tCDMp2j/ZQeeT8KND8v5KA6ad0t06xKyvxQ58VblCdlBWsHUF/AB6HtqAU0QlqhA3ZPewG7zuNM\n4jpex1xigKAunbggjXy62Wxh++wkYLjhAOYIXP+rsMFN1xdjd6/Ibv8Sed4fDoV9KQ==\n' class TAPRenderTest(trialhelpers.ArchiveTest): pass class SyncMetaTest(TAPRenderTest): """tests for non-doQuery sync queries. """ def testVersionRejected(self): """requests with bad version are rejected. """ return self.assertGETHasStrings(f"{_B}sync", { "REQUEST": "queryData", "VERSION": "2.1", "LAnG": "ADQL", "QUeRY": "ignored"}, ['', 'Version mismatch']) def testNoSyncPaths(self): """segments below sync are 404. """ return trialhelpers.runQuery(self.renderer, "GET", f"{_B}sync/foo", {} ).addCallback(lambda res: ddt #noflake: test instrumentation ).addErrback(lambda res: None) def testCapabilities(self): """simple get capabilities response looks ok. """ return self.assertGETHasStrings(f"{_B}sync", {"REQUEST": "getCapabilities"}, [ '']) def testTAPRootIsInfo(self): return self.assertGETHasStrings(f"{_B}", {}, [ ">ivo://x-testing/tap<", "The Unittest Suite's TAP end point."]) def testAvailability(self): return self.assertGETHasStrings(f"{_B}availability", {}, [ "true"]) def testCapabilitiesPath(self): return self.assertGETHasStrings(f"{_B}capabilities", {}, [ ' Unittest Suite's Table Access Protocol (TAP) service"]) def testExamples(self): return self.assertGETHasStrings(f"{_B}examples", {}, [ '']) class SyncQueryTest(TAPRenderTest): """tests for querying sync queries. """ aVOTable = os.path.join(base.getConfig("inputsdir"), "data/vizier_votable.vot") def testNoLangRejected(self): return self.assertGETHasStrings(f"{_B}sync", { "QUERY": 'SELECT alpha FROM test.adql WHERE alpha<3' }, [ ""]) def testBadLangRejected(self): return self.assertGETHasStrings(f"{_B}sync", { "LANG": "Furz", "QUERY": 'SELECT alpha FROM test.adql WHERE alpha<3' }, [ 'This service does' ' not support the query language Furz']) def testSimpleQuery(self): return self.assertGETHasStrings(f"{_B}sync", { "REQUEST": "doQuery", "LANG": "ADQL", "FORMAT": "text/xml", "QUERY": 'SELECT alpha, NULL as n FROM test.adql WHERE alpha<200' }, [ '', '2.0', 'NaN']) def testNameError(self): return self.assertGETHasStrings(f"{_B}sync", { "REQUEST": "doQuery", "LANG": "ADQL", "QUERY": 'SELECT doesnotexist FROM test.adql' }, [ 'INFO name="QUERY_STATUS" value="ERROR">Field query:' " No such field known: column 'doesnotexist' could not" ' be located in table metadata' ]) def testMultipleRequest(self): return self.assertGETHasStrings(f"{_B}sync", { "REQUEST": ["doQuery", "doQuery"], "LANG": "ADQL", "QUERY": 'SELECT alpha FROM test.adql WHERE alpha<2' }, [ '2 ']) def testNoOverflowWithTOP(self): return self.assertGETLacksStrings(f"{_B}sync", { "REQUEST": "doQuery", "LANG": "ADQL", "MAXREC": "1", "QUERY": 'SELECT TOP 1 alpha FROM test.adql' }, [ 'value="OVERFLOW"', ]) def testMAXREC0(self): return self.assertGETHasStrings(f"{_B}sync", { "REQUEST": "doQuery", "LANG": "ADQL", "MAXREC": "0", "QUERY": 'SELECT alpha FROM test.adql' }, [ 'Query successful', 'Field responseformat:' " 'xls' is not a valid value for responseformat"]) def testClearVOT(self): return self.assertGETHasStrings(f"{_B}sync", { "REQUEST": "doQuery", "LANG": "ADQL", "QUERY": 'SELECT CIRCLE(\'ICRS\', alpha, delta+1e-10, 10)' ' FROM test.adql WHERE alpha<3', "FORMAT": "votable/td" }, [ '2.0', '14.0', '10.0', 'xtype="circle"']) def testCSV(self): return self.assertGETHasStrings(f"{_B}sync", { "REQUEST": "doQuery", "LANG": "ADQL", "QUERY": 'SELECT alpha,delta FROM test.adql WHERE alpha<3', "FORMAT": "text/csv" }, [ '2.0,14.0']) def testTSV(self): return self.assertGETHasStrings(f"{_B}sync", { "REQUEST": "doQuery", "LANG": "ADQL", "QUERY": 'SELECT alpha, delta FROM test.adql WHERE alpha<3', "FORMAT": "TSV" }, [ '2.0\t14.0']) def testJSON(self): return self.assertGETHasStrings(f"{_B}sync", { "REQUEST": "doQuery", "LANG": "ADQL", "QUERY": "SELECT TOP 3 column_name, datatype FROM TAP_SCHEMA.columns" " WHERE table_name='tap_schema.tables' ORDER BY column_name", "RESPONSEFORMAT": "application/json", "MAXREC": "1" }, [ '"contains": "table"', '"description": "ADQL datatype"', # DaCHS buglet: maxrec=1 returns two rows. '["description", "unicodeChar"], ["nrows", "long"]', '"queryStatus": "OVERFLOW"']) def testBin2Table(self): return self.assertGETHasStrings(f"{_B}sync", { "REQUEST": "doQuery", "LANG": "ADQL", "QUERY": 'SELECT alpha, delta FROM test.adql WHERE alpha<3', "FORMAT": "application/x-votable+xml;serialization=BINARY2" }, [ 'BINARY2', 'AEAAAABBYAAA']) def testFITSTable(self): return self.assertGETHasStrings(f"{_B}sync", { "REQUEST": "doQuery", "LANG": "ADQL", "QUERY": 'SELECT alpha, delta FROM test.adql WHERE alpha<3', "RESPONSEFORMAT": "fits", "MAXREC": "0", }, [ 'SIMPLE =', "TYPE1 = 'alpha '", "TCOMM1 = 'A sample RA'"]) def testBadUploadSyntax(self): return self.assertPOSTHasStrings(f"{_B}sync", { "REQUEST": "doQuery", "UPLOAD": "bar", "LANG": "ADQL", "QUERY": 'SELECT * FROM test.adql' }, [ "only allow regular SQL identifiers"]) def testBadUploadSyntax2(self): return self.assertPOSTHasStrings(f"{_B}sync", { "REQUEST": "doQuery", "UPLOAD": "bar,http://x.y;", "LANG": "ADQL", "QUERY": 'SELECT * FROM test.adql'}, [ "only allow regular SQL identifiers" ]) def testNonExistingUpload(self): return self.assertPOSTHasStrings(f"{_B}sync", { "REQUEST": "doQuery", "UPLOAD": "bar,http://127.0.0.1:65000", "LANG": "ADQL", "QUERY": 'SELECT * FROM test.adql' }, [ "'http://127.0.0.1:65000' cannot be retrievedField UPLOAD:' ' Syntax error in UPLOAD parameter' ]) def testInlineUploadFromArgsWorks(self): with open(self.aVOTable, "rb") as f: uploadFile = trialhelpers.FakeFile("../Hägar/klause", f.read()) return self.assertPOSTHasStrings(f"{_B}sync", { "REQUEST": "doQuery", "UPLOAD": "bar,param:HoNk", "LANG": "ADQL", "QUERY": 'SELECT * FROM tap_upload.bar join test.adql on' ' (alpha-"_RAJ2000"<0)', "HoNk": uploadFile, }, [ 'xmlns="http://www.ivoa.net/xml/VOTable/', 'ucd="pos.eq.ra;meta.main"', 'encoding="base64"']) def testSensibleErrorOnMalformedUpload(self): return self.assertPOSTHasStrings(f"{_B}sync", { "REQUEST": "doQuery", "UPLOAD": "bar,param:HoNk", "LANG": "ADQL", "QUERY": 'SELECT * FROM tap_upload.bar join test.adql on' ' (alpha-"_RAJ2000"<0)', "HoNk": trialhelpers.FakeFile("../Hägar/klause", b"""
13
break here
"""), }, [ "While ingesting upload bar: Invalid literal for double (field x):" " 'break here'"]) def testMissingInlineParameter(self): return self.assertPOSTHasStrings(f"{_B}sync", { "REQUEST": "doQuery", "UPLOAD": "bar,param:HoNk", "LANG": "ADQL", "QUERY": 'SELECT top 1 * FROM tap_upload.bar', "MoNk": ("upl", open(self.aVOTable, "rb")), }, [ '' "Field UPLOAD: No inline upload 'HoNk' found"]) def testWithUploadOldGeom(self): return self.assertPOSTHasStrings(f"{_B}sync", { "REQUEST": "doQuery", "UPLOAD": "t1,param:HoNk", "LANG": "ADQL", "FORMAT": "votable/td", "QUERY": "SELECT ssa_location," " POINT('ICRS', COORD1(ssa_location), COORD2(ssa_location)) AS np" " FROM test.adql join tap_upload.t1" " ON 1=contains(ssa_location, circle('', alpha, delta, 0.002))", "HoNk": trialhelpers.FakeFile( "upl", utils.getDirtyBytes(votWithOldGeom)) }, [ "Position UNKNOWNFrame 2.0009", "14.00100", "2.00", " 14.00"]) def testWithUploadGeom(self): return self.assertPOSTHasStrings(f"{_B}sync", { "REQUEST": "doQuery", "UPLOAD": "t1,param:HoNk", "LANG": "ADQL", "FORMAT": "votable/td", "QUERY": 'SELECT * FROM test.adql join tap_upload.t1' " on 1=contains(ssa_location, circle('', alpha, delta, 0.002))", "HoNk": trialhelpers.FakeFile("upl", utils.getDirtyBytes(votWithDALIGeom)), }, ["2.00", " 14.00"]) def testArraySerialisation(self): return self.assertPOSTHasStrings(f"{_B}sync", { "REQUEST": "doQuery", "LANG": "ADQL", "QUERY": 'SELECT gavo_histogram(alpha, 0, 360, 4) as h FROM test.adql'}, [ 'AAAABgAAAAAAAAACAAAAAAAAAAAAAAABAAAAAA==', 'arraysize="*" datatype="int" name="h"']) class SimpleAsyncTest(TAPRenderTest): """tests for some non-ADQL async queries. """ def testAsyncDirRedirects(self): def assertLocation(result): self.assertTrue(result[1].getLocationValue().endswith(f"{_B}async")) return result return self.assertStatus(f"{_B}async/", 302 ).addCallback(assertLocation) def testVersionRejected(self): return self.assertPOSTHasStrings(f"{_B}async", { "REQUEST": "doQuery", "LANG": "ADQL", "VERSION": "0.1"}, ['', 'Version mismatch; this']) def testJobList(self): return self.assertGETHasStrings(f"{_B}async", {}, [ 'UWS job \'23\' could not']) def testLifecycle(self): """tests job creation, redirection, phase, and deletion. """ # This one is too huge and much too slow for a unit test. Still # I want at least one integration-type test in here since the # big test probably won't be run at every commit. def assertDeleted(result, jobId): self.assertEqual(result[1].code, 303) next = result[1].getLocationValue() self.assertTrue(next.endswith(f"{_B}async"), "Deletion redirect doesn't point to job list but to %s"%next) return self.assertGETLacksStrings(next, {}, ['jobref id="%s"'%jobId] ).addCallback(lambda res: reactor.disconnectAll()) def delete(ignored, jobId): return trialhelpers.runQuery(self.renderer, "DELETE", f"{_B}async/"+jobId, {} ).addCallback(assertDeleted, jobId) def checkResult(ignored, jobId): return self.assertGETHasStrings( f"{_B}async/%s/results/result"%jobId, {}, [ 'name="sql_query"', 'QAAAAA=='] ).addCallback(delete, jobId) def waitForCompletion(result, jobId): if b"EXECUTING" in result[0]: # wait for completion once more return self.assertGETHasStrings(f"{_B}async/%s"%jobId, {"WAIT": "10"}, ["COMPLETED"] ).addCallback(checkResult, jobId) else: self.assertTrue(b"COMPLETED" in result[0], "'%s' doesn't have COMPLETED in it") return checkResult(result, jobId) def assertStarted(lastRes, jobId): # lastRes must be a redirect to the job info page req = lastRes[1] self.assertEqual(req.code, 303) self.assertEqual(req.getLocationValue(), "http://localhost:8080/__system__/tap/run/async/"+jobId) return self.assertGETHasStrings(f"{_B}async/%s"%jobId, {"WAIT": "10"}, [""]).addCallback(waitForCompletion, jobId) def promote(ignored, jobId): return trialhelpers.runQuery(self.renderer, "POST", f"{_B}async/%s/phase"%jobId, {"PHASE": "RUN"} ).addCallback(assertStarted, jobId) def checkPlan(ignored, jobId): return self.assertGETHasStrings(f"{_B}async/%s/plan"%jobId, {}, ["Seq Scan on adql", "alpha < '3'::double"] ).addCallback(promote, jobId) def checkQuote(ingored, jobId): return self.assertGETHasStrings(f"{_B}async/%s/quote"%jobId, {}, ['-'] ).addCallback(checkPlan, jobId) def checkPhase(ignored, jobId): return self.assertGETHasStrings(f"{_B}async/%s/phase"%jobId, {}, ['PENDING'] ).addCallback(checkQuote, jobId) def checkJobInfo(jobId): return self.assertGETHasStrings(f"{_B}async/%s"%jobId, {}, [ '5: raise time.sleep(0.05) except ChildProcessError: # I don't think these should come with WNOHANG, but... well # there they are, and since we assert the child isn't there, # I'm happy with them, too. break else: break return result def destroy(result, jobId): job = tap.WORKER_SYSTEM.getJob(jobId) self.assertEqual(job.phase, "EXECUTING") wd = job.getWD() pid = job.pid return trialhelpers.runQuery(self.renderer, "DELETE", f"/tap/async/{jobId}", {}, ).addCallback(assertDeleted, jobId, pid, wd) def waitUntilRuns(result, jobId): return self.assertGETHasStrings(f"tap/async/{jobId}", {"WAIT": "10"}, ['"query">JUST HANG']).addCallback(destroy, jobId) def startJob(result): jobURL = result[1].getLocationValue() self.assertTrue(re.search("tap/run/async", jobURL)) jobId = jobURL.split("/")[-1] return self.assertPOSTHasStrings(f"/tap/async/{jobId}/phase", {"PHASE": "RUN"}, ["Unittest Suite -- See Other"] ).addCallback(waitUntilRuns, jobId) return self.assertPOSTHasStrings("tap/async", { "LANG": "ADQL", "QUERY": "JUST HANG around"}, ["Unittest Suite -- See Other"] ).addCallback(startJob) class PersistentUploadTest(TAPRenderTest): with open( base.getConfig("inputsDir") / "data" / "vizier_votable.vot", "rb") as f: someVOT = io.BytesIO(f.read()) def assert404Status(self, res): self.assertEqual(res[1].code, 404) return res def testNoTable(self): return self.assertGETHasStrings("tap/user_tables", {}, [ 'name="QUERY_STATUS"', "No persistent tables listing for anonymous users.<",]) def testTablesListing(self): return self.assertGETHasStrings("tap/user_tables", {}, [ "tap_user", "A schema containing users' uploads"], rm=trialhelpers.addCreds) def testBadTableName(self): return self.assertGETHasStrings("tap/user_tables/666", {}, [ 'name="QUERY_STATUS"', '>You must give a valid upload table name here<'] ).addCallback(self.assert404Status) def testNoPosting(self): return self.assertPOSTHasStrings("tap/user_tables/test_upload", {}, [ 'name="QUERY_STATUS"', "This resource cannot respond to the HTTP 'POST' method"]) def testNoMediaType(self): def makePUT(request): request.setPayload(self.someVOT, "application/octet-stream") request.method = "PUT" return self.assertPOSTHasStrings( f"{_B}user_tables/test_upload", {}, [ "Do not know how to ingest application/octet-stream tables<", "if in doubt, use application/x-votable+xml" ], rm=makePUT) def testSuccessfulUpload(self): def makePUT(request): request.setPayload(self.someVOT, base.votableType) request.method = "PUT" def makeDELETE(request): request.method = "DELETE" def assertCleanedUp(result): self.assertEqual(result[0], b'Dropped user table test_upload\n') self.assertEqual(result[1].code, 200) with base.getTableConn() as conn: self.assertEqual(list(conn.query( "select user_name, table_name" " from tap_user.tables where user_name='anonymous' and" " table_name='test_upload'")), []) self.assertEqual(base.UnmanagedQuerier(conn).getTableType( "tap_user._anonymous_test_upload"), None) return result def tryDELETE(result): return self.assertGETHasStrings( f"{_B}user_tables/test_upload", {}, [ "Dropped user table test_upload"], rm=makeDELETE ).addCallback(assertCleanedUp) def tryGET(result): self.assertEqual(result[1].code, 200) with base.getTableConn() as conn: self.assertEqual( list(conn.query("select color from tap_user._anonymous_test_upload" " where tyc2='1265-01176-1'")), [(0.25,)]) return self.assertGETHasStrings( f"{_B}user_tables/test_upload", {}, [ "tap_user.test_upload", "tyc2"] ).addCallback(tryDELETE) return self.assertPOSTHasStrings( f"{_B}user_tables/test_upload", {}, [ "Query this table as tap_user.test_upload", ], rm=makePUT ).addCallback(tryGET) def testCreateAs(self): def makeDELETE(request): request.method = "DELETE" def cleanup(res): return self.assertGETHasStrings(f"{_B}user_tables/fromquery", {}, [ "Dropped user table fromquery"], rm=makeDELETE) def tryQuery(res): return self.assertGETHasStrings( f"{_B}sync", {"LANG": "ADQL", "QUERY": "select * from tap_user.fromquery"}, [ 'QAAAAA==']) return self.assertGETHasStrings( f"{_B}sync", {"LANG": "ADQL", "QUERY": "CREATE TABLE tap_user.fromquery AS" " SELECT alpha FROM test.adql WHERE alpha<3"}, [ 'A sample RA', # metadata present '', # empty VOTable ]).addCallback(tryQuery ).addBoth(cleanup) def testNoReservedWords(self): def makePUT(request): request.setPayload(self.someVOT, base.votableType) request.method = "PUT" return self.assertPOSTHasStrings( f"{_B}user_tables/table", {}, [ 'value="ERROR"', "'table' cannot be used as an upload table name (which must be" " regular ADQL identifiers, in particular not ADQL reserved words)." ], rm=makePUT) atexit.register(trialhelpers.provideRDData("test", "ADQLTest"))