""" Tests for our datalink implementation. Note that there's also sodatest, which should contain tests for DaCHS-defined meta makers and data functions (at least where concerned with parameters contained in SODA). """ #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 gc from gavo.helpers import testhelpers from gavo import api from gavo import svcs from gavo import votable from gavo.helpers import trialhelpers from gavo.protocols import datalink import tresc class MetaErrorSemanticsInferenceTest(testhelpers.VerboseTest): def _getDatalinkErrors(self, metaMakerDef): core = api.parseFromString(datalink.DatalinkCore, """ return ProductDescriptor("contrived", "contrived", None, "application/bogus") %s """%metaMakerDef) args = {"ID": "contrived"} #noflake: variable stolen by dl core dlCore = core.adaptForRenderer( svcs.getRenderer("dlmeta"), svcs.emptyQueryMeta) _, _, errors = dlCore.getMetaForDescriptor(dlCore.descriptors[0]) return errors def testSyntaxErrorInferred(self): errors = self._getDatalinkErrors(""" return return yield descriptor.makeLink("heu") """) self.assertEqual(errors[0].semantics, "#grassy") def testDefaultError(self): errors = self._getDatalinkErrors(""" return return yield descriptor.makeLink("heu") """) self.assertEqual(errors[0].semantics, "http://dc.g-vo.org/datalink#other") class _DumbSODAService(testhelpers.TestResource): resources = [("prodtestTable", tresc.prodtestTable)] def make(self, dependents): svc = api.parseFromString(svcs.Service, """ """) svc.parent = testhelpers.getTestRD() return svc class DLInterfaceTest(testhelpers.VerboseTest): resources = [("svc", _DumbSODAService()), ("ssa", tresc.ssaTestTable)] def testIDNecessary(self): self.assertRaisesWithMsg(api.ValidationError, "Field ID: ID is mandatory with dlget", trialhelpers.runSvcWith, (self.svc, "dlget", {})) def testMissingAccref(self): tree = testhelpers.getXMLTree( trialhelpers.runSvcWith(self.svc, "dlmeta", {"ID": "junkjunk"})[1]) self.assertEqual(tree.xpath("//TR/TD[4]")[0].text, "NotFoundFault: accref 'junkjunk' could" " not be located in product table") def testWrongAuthority(self): tree = testhelpers.getXMLTree( trialhelpers.runSvcWith(self.svc, "dlmeta", {"ID": "ivo://junkjunk/knatter"})[1]) self.assertEqual(tree.xpath("//TR/TD[4]")[0].text, "NotFoundFault: The authority in the dataset identifier" " 'ivo://junkjunk/knatter' could not be located in the" " authorities managed here") def testNoNoArgProc(self): tree = testhelpers.getXMLTree( trialhelpers.runSvcWith(self.svc, "dlmeta", {"ID": "data/spec3.ssatest.vot"})[1], debug=False) # there's do inputKey-generating metaMaker here, so there shouldn't # be a #proc line or a meta resource self.assertEqual(len(tree.xpath("//TD[.='#proc']")), 0) self.assertEqual(len(tree.xpath("//RESOURCE")), 1) class _MetaMakerTestData(testhelpers.TestResource): # test data for datalink metadata generation resources = [ ("prodtestTable", tresc.prodtestTable)] def make(self, dependents): svc = api.parseFromString(svcs.Service, """ data A sample processing service def addExtras(descriptor): descriptor.bogosity = len(descriptor.pubDID) ik = MS(InputKey, name="format", type="text", ucd="meta.format", description="Output format desired", values=MS(Values, options=[MS(Option, content_=descriptor.mime), MS(Option, content_="application/fits")])) ik.setProperty("defaultForForm", "application/fits") yield ik yield descriptor.makeLink("http://foo/bar", contentType="test/junk", contentLength=500002, localSemantics="geneigt", contentQualifier="#other") yield descriptor.makeLink("http://foo/baz", contentType="test/gold", semantics="#calibration") if descriptor.pubDID.endswith("b.imp"): yield DatalinkFault.NotFoundFault("ivo://not.asked.for", "Cannot locate other mess") yield descriptor.makeLinkFromFile( "no.such.file", description="An unrelated nothing") yield descriptor.makeLinkFromFile( "expected.missing", description="A file that just does not" " exist and should not create diagnostics.", semantics="http://www.g-vo.org/dl#mustnothappen", suppressMissing=True) yield descriptor.makeLinkFromFile("data/map1.map", description="Some mapping", semantics="http://www.g-vo.org/dl#related", contentQualifier="#something") if descriptor.pubDID!='ivo://x-testing/~?data/a.imp': return yield ExternalProcLinkDef( descriptor.pubDID, [ MS(InputKey, name="QUERY", type="text", content_="SELECT * FROM some_table_name"), MS(InputKey, name="LANG", type="text", content_="ADQL")], "http://dc.g-vo.org/tap/sync", "GAVO TAP", "TAP service with an alternative representation of the data", standardId="ivo://ivoa.net/std/tap") yield descriptor.makeLink( f"http://bogosity.code/{descriptor.bogosity}") """) svc.parent = testhelpers.getTestRD() mime, data = trialhelpers.runSvcWith(svc, "dlmeta", { "id": [ api.getStandardPubDID("data/a.imp"), api.getStandardPubDID("data/b.imp"), ]}) from gavo.registry import capabilities capEl = capabilities.getCapabilityElement(svc.publications[0]) return (mime, testhelpers.getXMLTree(data, debug=False), list(next(votable.parseBytes(data))), testhelpers.getXMLTree(capEl.render(), debug=False)) _metaMakerTestData = _MetaMakerTestData() class DatalinkMetaMakerTest(testhelpers.VerboseTest): resources = [("serviceResult", _metaMakerTestData), ("prodtestTable", tresc.prodtestTable)] def testServiceMetaDeclared(self): infoEl = self.serviceResult[1].xpath( "RESOURCE[@type='results']/INFO[@name='reference_url']")[0] self.assertEqual( infoEl.get("value"), "http://localhost:8080/data/test/foo/info") def testStandardId(self): infoEl = self.serviceResult[1].xpath( "RESOURCE[@type='results']/INFO[@name='standardID']")[0] self.assertTrue(infoEl.text.startswith("Written by DaCHS")) self.assertEqual( infoEl.get("value"), "ivo://ivoa.net/std/datalink#links-1.1") def testIDSet(self): svc1 = self.serviceResult[1].xpath("//RESOURCE[@utype='adhoc:service']")[0] self.assertEqual( svc1.xpath("GROUP/PARAM[@name='ID']")[0].get("value"), "ivo://x-testing/~?data/a.imp") svc2 = self.serviceResult[1].xpath("//RESOURCE[@utype='adhoc:service']")[1] self.assertEqual( svc2.xpath("GROUP/PARAM[@name='ID']")[0].get("value"), "ivo://x-testing/~?data/b.imp") def testMimeOk(self): self.assertEqual(self.serviceResult[0], "application/x-votable+xml;content=datalink") def testUCDPresent(self): tree = self.serviceResult[1] self.assertEqual( tree.xpath("//PARAM[@name='format']")[0].get("ucd"), "meta.format") def testTypeTranslationWorks(self): tree = self.serviceResult[1] self.assertEqual( tree.xpath("//PARAM[@name='format']")[0].get("arraysize"), "*") def testPreset(self): link = self.serviceResult[1].xpath("//PARAM[@name='format']/LINK")[0] self.assertEqual(link.get("content-role"), "#pre-set") self.assertEqual(link.get("value"), "application/fits") def testOptionsRepresented(self): tree = self.serviceResult[1] self.assertEqual( tree.xpath("//PARAM[@name='format']/VALUES/OPTION")[0].get("value"), "text/plain") self.assertEqual( tree.xpath("//PARAM[@name='format']/VALUES/OPTION")[1].get("value"), "application/fits") def testAccessURLPresent(self): tree = self.serviceResult[1] self.assertEqual( tree.xpath("//PARAM[@name='accessURL']")[0].get("value"), "http://localhost:8080/data/test/foo/dlget") def testAddExtras(self): rows = self.serviceResult[1].xpath("//TR[TD[6]='#testing']") self.assertEqual(rows[0][1].text, "http://bogosity.code/28") self.assertEqual(rows[1][1].text, "http://bogosity.code/28") def testExternalProc(self): tree = testhelpers.pickSingle( self.serviceResult[1].xpath("RESOURCE[@utype='adhoc:service'][" "INFO[@name='title']/@value='GAVO TAP']")) self.assertEqual( testhelpers.pickSingle(tree.xpath("PARAM[@name='standardID']") ).get("value"), "ivo://ivoa.net/std/tap") self.assertEqual( testhelpers.pickSingle(tree.xpath("PARAM[@ucd='meta.ref.url']") ).get("value"), "http://dc.g-vo.org/tap/sync") self.assertEqual( testhelpers.pickSingle(tree.xpath("GROUP/PARAM[@name='QUERY']") ).get("value"), "SELECT * FROM some_table_name") refLine = self.serviceResult[1].xpath("//TR[TD='TAP service with an alternative representation of the data']")[0] self.assertEqual( refLine[2].text, # svc_id tree.get("ID")) def testCapability(self): intfEl = self.serviceResult[3].xpath("//interface")[0] self.assertEqual( intfEl.attrib["{http://www.w3.org/2001/XMLSchema-instance}type"], "vs:ParamHTTP") self.assertEqual(intfEl.xpath("queryType")[0].text, "GET") self.assertEqual(intfEl.xpath("resultType")[0].text, 'application/x-votable+xml;content=datalink') self.assertEqual(intfEl.xpath("accessURL")[0].text, 'http://localhost:8080/data/test/foo/dlmeta') self.assertEqual(self.serviceResult[3].xpath("/capability")[0].attrib[ "standardID"], "ivo://ivoa.net/std/DataLink#links-1.1") def testServiceTitle(self): # the title info is used by the datalink XSLT self.assertEqual(self.serviceResult[1].xpath( "RESOURCE[@ID='proc_svc']/INFO[@name='title']")[0].get("value"), "Processing service with id foo") def testCapabilityParameters(self): intfEl = self.serviceResult[3].xpath("//interface")[0] for el in intfEl.xpath("param"): parName = el.xpath("name")[0].text if parName=="ID": self.assertEqual(el.attrib["std"], "true") self.assertEqual(el.xpath("ucd")[0].text, "meta.id;meta.main") elif parName=="responseformat": datatype = el.xpath("dataType")[0] self.assertEqual(datatype.text, "char") self.assertEqual(datatype.get("arraysize"), "*") elif parName=="maxrec": self.assertEqual(el.xpath("description")[0].text, "Maximum number of records returned." " Pass 0 to retrieve service parameters.") elif parName=="verb": pass # forget it; perhaps we'll use it some day. else: raise AssertionError("Unexpected Parameter %s"%parName) def testAsyncDeclared(self): svc = api.parseFromString(svcs.Service, """ yield MS(InputKey, name="PAR", type="text") """) svc.parent = testhelpers.getTestRD() mime, data = trialhelpers.runSvcWith(svc, "dlmeta", { "ID": [ api.getStandardPubDID("data/a.imp"), ]}) tree = testhelpers.getXMLTree(data, debug=False) self.assertEqual(len(tree.xpath("//TR")), 4) self.assertEqual( set(['http://localhost:8080/data/test/foo/dlget', 'http://localhost:8080/data/test/foo/dlasync']), set([p.get("value") for p in tree.xpath("//PARAM[@name='accessURL']")])) self.assertEqual( set(['ivo://ivoa.net/std/soda#async-1.0', 'ivo://ivoa.net/std/soda#sync-1.0']), set([p.get("value") for p in tree.xpath("//PARAM[@name='standardID']")])) def testCoreForgetting(self): from gavo.svcs import renderers args = {"ID": api.getStandardPubDID("data/ex.fits")} svc = api.getRD("data/cores").getById("dl") renderer = renderers.getRenderer("dlmeta") gns = testhelpers.getMemDiffer() core = svc.core.adaptForRenderer(renderer, svcs.emptyQueryMeta) class _Sentinel(object): pass s = _Sentinel() core.ref = s coreId = id(core.__dict__) self.assertTrue(coreId in set(id(r) for r in gc.get_referrers(s))) it = svcs.CoreArgs.fromRawArgs(core.inputTable, args) core.runForMeta(svc, it, svcs.emptyQueryMeta) core.finalize() core.inputTable.breakCircles() del core del it gc.collect() # Hack to work around delayed removal of core from locals() # perhaps that's actually a python 3.7 bug? if "core" in locals(): locals().pop("core") ns = gns() self.assertEqual(len(ns), 0, "Uncollected garbage: %s"%ns) self.assertFalse(coreId in set(id(r) for r in gc.get_referrers(s)), "core still lives and references s") class _MetaMakerTestRows(testhelpers.TestResource): resources = [ ("serviceResult", _metaMakerTestData)] def make(self, dependents): td = api.resolveCrossId("//datalink#dlresponse", None) rows = {} for tuple in dependents["serviceResult"][2]: row = td.makeRowFromTuple(tuple) rows.setdefault((row["ID"], row["semantics"]), []).append(row) return rows class DatalinkMetaRowsTest(testhelpers.VerboseTest): resources = [("rows", _MetaMakerTestRows()), ("serviceResult", _metaMakerTestData)] def testAllLinks(self): self.assertEqual(len(self.rows), 17) for key, r in list(self.rows.items()): if key[1]!="#proc": self.assertEqual(len(r), 1) def testAllWithId(self): self.assertEqual(set(r[0] for r in self.rows), set(['ivo://x-testing/~?data/b.imp', 'ivo://x-testing/~?data/a.imp', 'ivo://not.asked.for'])) def testAccessURLStatic(self): self.assertEqual(self.rows[ ('ivo://x-testing/~?data/b.imp', '#sibling')][0]["access_url"], 'http://foo/bar') def testAccessURLAccess(self): self.assertEqual(self.rows[ ('ivo://x-testing/~?data/b.imp', '#proc')][0]["access_url"], # take the actual access URL from the service block. None) def testAccessURLSelf(self): self.assertEqual(self.rows[ ('ivo://x-testing/~?data/b.imp', '#this')][0]["access_url"], "http://localhost:8080/getproduct/data/b.imp") self.assertEqual(self.rows[ ('ivo://x-testing/~?data/a.imp', '#this')][0]["access_url"], "http://localhost:8080/getproduct/data/a.imp") def testProcDescribed(self): self.assertEqual(self.rows[ ('ivo://x-testing/~?data/b.imp', '#proc')][0]["description"], "A sample processing service") def testMimes(self): self.assertEqual(self.rows[('ivo://x-testing/~?data/a.imp', '#calibration')][0]["content_type"], 'test/gold') def testSemantics(self): self.assertEqual(set(r[1] for r in self.rows), set(['#proc', '#this', '#sibling', '#calibration', '#preview', "#testing", "http://dc.g-vo.org/datalink#other", 'http://www.g-vo.org/dl#related', 'http://www.g-vo.org/dl#unrelated', ])) def testLocalSemantics(self): withLS = self.rows[('ivo://x-testing/~?data/b.imp', '#sibling')] self.assertEqual(withLS[0]["local_semantics"], "geneigt") def testSizes(self): self.assertEqual(self.rows[('ivo://x-testing/~?data/a.imp', '#sibling')][0]["content_length"], 500002) self.assertEqual(self.rows[('ivo://x-testing/~?data/a.imp', '#calibration')][0]["content_length"], None) def testServiceLink(self): svcRow = self.rows[('ivo://x-testing/~?data/a.imp', '#proc')][0] resId = svcRow["service_def"] for res in self.serviceResult[1].xpath("//RESOURCE"): if res.attrib.get("ID")==resId: break else: self.fail("Processing service not in datalink links") self.assertEqual(res.attrib.get("type"), "meta") self.assertEqual(res.attrib.get("utype"), "adhoc:service") def testSelfMeta(self): selfRow = self.rows[('ivo://x-testing/~?data/b.imp', '#this')][0] self.assertEqual(selfRow["content_type"], "text/plain") self.assertEqual(selfRow["content_length"], 73) def testMetaError(self): errors = self.rows[('ivo://not.asked.for', datalink.DEFAULT_SEMANTICS)] self.assertEqual(errors[0]["error_message"], 'NotFoundFault: Cannot locate other mess') def testPreviewMetaURL(self): previewRow = self.rows[('ivo://x-testing/~?data/b.imp', '#preview')][0] self.assertEqual(previewRow["access_url"], "http://example.com/borken.jpg") self.assertEqual(previewRow["content_type"], "image/jpeg") def testPreviewMetaAuto(self): previewRow = self.rows[('ivo://x-testing/~?data/a.imp', '#preview')][0] self.assertEqual(previewRow["access_url"], "http://localhost:8080/getproduct/data/a.imp?preview=True") self.assertEqual(previewRow["content_type"], "text/plain") def testFromNonExistingFile(self): errRow = self.rows[('ivo://x-testing/~?data/b.imp', 'http://www.g-vo.org/dl#unrelated')][0] self.assertEqual(errRow["error_message"], "NotFoundFault: No file for linked item") self.assertEqual(errRow["description"], "An unrelated nothing") def testExpectedNonExisting(self): self.assertFalse(('ivo://x-testing/~?data/b.imp', 'http://www.g-vo.org/dl#mustnothappen') in self.rows, "suppressMissing in makeLinkFromFile broken?") def testFromFile(self): row = self.rows[('ivo://x-testing/~?data/b.imp', 'http://www.g-vo.org/dl#related')][0] self.assertEqual(row["error_message"], None) self.assertEqual(row["content_length"], 8) self.assertEqual(row["description"], "Some mapping") self.assertEqual(row["content_type"], "application/octet-stream") def testSemanticsOverridable(self): row = self.rows[('ivo://x-testing/~?data/b.imp', "#calibration")][0] self.assertEqual(row["content_type"], "test/gold") def contentQualifierDefault(self): row = self.rows[('ivo://x-testing/~?data/b.imp', "#calibration")][0] self.assertEqual(row["content_qualifier"], None) def contentQualifierFromLink(self): row = self.rows[('ivo://x-testing/~?data/a.imp', "#sibling")][0] self.assertEqual(row["content_qualifier"], "#other") def contentQualifierFromFile(self): row = self.rows[('ivo://x-testing/~?data/a.imp', "http://www.g-vo.org/dl#related")][0] self.assertEqual(row["content_qualifier"], "#something") class _TimestampService(testhelpers.TestResource): def make(self, ignored): svc = api.parseFromString(svcs.Service, """ A Test Service yield MS(InputKey, name="t", type="timestamp[2]", ucd="time.epoch", xtype="interval", values=MS(Values, default=[ datetime.datetime(2020, 1, 1),datetime.datetime(2021, 1, 1)])) if pubDID=="bogus": return DatalinkFault.NotFoundFault(pubDID, "No bogus data here") res = ProductDescriptor(pubDID, "bogus", None, "application/bogus") res.contentQualifier = "#timeseries" return res assert isinstance(args["t"][0], datetime.datetime) assert isinstance(args["t"][1], datetime.datetime) descriptor.data = "%s: %s"%(descriptor.pubDID, repr(args["t"])) """) svc.parent = testhelpers.getTestRD() return svc class TimestampTest(testhelpers.VerboseTest): resources = [("svc", _TimestampService())] def testTimestampsMeta(self): mime, data = trialhelpers.runSvcWith(self.svc, "dlmeta", { "ID": "beknackt"}) tree = testhelpers.getXMLTree(data, debug=False) # this sucks -- xtype=interval and xtype=timestamp is mutually # exclusive at this point. I invent something. pe = tree.xpath("//GROUP[@name='inputParams']/PARAM[@name='t']") param = testhelpers.pickSingle(pe) self.assertEqual(param.get("xtype"), "timestamp-interval") self.assertEqual(param.get("arraysize"), "19x2") self.assertEqual(param.get("datatype"), "char") self.assertEqual( param.get("value"), "2020-01-01T00:00:002021-01-01T00:00:00") def testTimestampsParsed(self): data = trialhelpers.runSvcWith(self.svc, "dlget", { "ID": "beknackt", "t": "2008-04-01T12:00:002008-05-01T12:00:00"} ) self.assertEqual(data, # should be different when we properly do the xtype thing "beknackt: [datetime.datetime(2008, 4, 1, 12, 0), datetime.datetime(2008, 5, 1, 12, 0)]") def testFaultsFromDescGen(self): data = trialhelpers.runSvcWith(self.svc, "dlmeta", { "ID": "bogus"}) s, metadata = votable.loads(data[1]) res = list(metadata.iterDicts(s)) self.assertEqual( (testhelpers.pickSingle(res)["ID"], res[0]["error_message"], res[0]["semantics"]), ("bogus", "NotFoundFault: No bogus data here", "#this")) def testTitleMetaInInfo(self): data = trialhelpers.runSvcWith(self.svc, "dlmeta", { "ID": "beknackt"})[1] tree = testhelpers.getXMLTree(data, debug=False) self.assertEqual(tree.xpath( "RESOURCE[@ID='proc_svc']/INFO[@name='title']")[0].get("value"), "A Test Service") def testThisMeta(self): data = trialhelpers.runSvcWith(self.svc, "dlmeta", { "ID": "tictac"})[1] tree = testhelpers.getXMLTree(data, debug=False) # semantics in TD 8, contentQualifier 10 self.assertEqual(tree.xpath( "//TR[TD[6]='#this']/TD[10]")[0].text, "#timeseries") class _TwoDatalinkServices(testhelpers.TestResource): def make(self, ignored): rd = api.parseFromString(api.RD, """ res = ProductDescriptor(pubDID, None, None, None) res.suppressAutoLinks = True return res yield MS(InputKey, name="arg1", type="text", description="First service's argument 1") yield ProcLinkDef(descriptor.pubDID, rd.getById("dl2")) yield ProcLinkDef(descriptor.pubDID, rd.getById("dl3")) An external datalink service yield MS(InputKey, name="knurr", type="text", description="Second service's argument") Another external datalink service yield MS(InputKey, name="knurr", type="text", description="One of the third service's arguments") yield MS(InputKey, name="ruck", type="integer", description="an integer-valued argument") """) metaTree = testhelpers.getXMLTree( trialhelpers.runSvcWith(rd.getById("dl"), "dlmeta", {"ID": "junkjunk"})[1], debug=False) return metaTree, rd class SecondServiceTest(testhelpers.VerboseTest): resources = [("treeAndRD", _TwoDatalinkServices())] def testProcLinks(self): svc1, svc2, svc3 = self.treeAndRD[0].xpath("//TR[TD[6]='#proc']") self.assertEqual(svc1.xpath("TD[6]")[0].text, "#proc") self.assertEqual(svc1.xpath("TD[1]")[0].text, "junkjunk") self.assertEqual(svc1.xpath("TD[3]")[0].text, "ext_svc") self.assertEqual(svc1.xpath("TD[5]")[0].text, "An external datalink service") self.assertEqual(svc2.xpath("TD[6]")[0].text, "#proc") self.assertEqual(svc2.xpath("TD[3]")[0].text, "ext_svc-02") self.assertEqual(svc2.xpath("TD[5]")[0].text, "Another external datalink service") self.assertEqual(svc3.xpath("TD[6]")[0].text, "#proc") self.assertEqual(svc3.xpath("TD[3]")[0].text, "proc_svc") self.assertEqual(svc3.xpath("TD[5]")[0].text, "An interactive service on this dataset.") def testBuiltinDef(self): svcrsc = self.treeAndRD[0].xpath("RESOURCE[@ID='proc_svc']")[0] self.assertEqual(svcrsc.xpath("DESCRIPTION")[0].text, "An interactive service on this dataset.") self.assertEqual(svcrsc.get("utype"), "adhoc:service") self.assertEqual(svcrsc.get("name"), "dl") self.assertEqual(svcrsc.xpath("PARAM[@name='standardID']")[0] .get("value"), "ivo://ivoa.net/std/soda#sync-1.0") self.assertEqual(svcrsc.xpath("PARAM[@name='accessURL']")[0] .get("value"), "http://localhost:8080/temporary/dl/dlget") self.assertEqual(svcrsc.xpath("GROUP/PARAM[@name='arg1']/DESCRIPTION" )[0].text, "First service's argument 1") def testExt1Rsc(self): svcrsc = self.treeAndRD[0].xpath("RESOURCE[@ID='ext_svc']")[0] self.assertEqual(svcrsc.get("name"), "dl2") self.assertEqual(svcrsc.xpath("GROUP/PARAM[@name='ID']")[0] .get("value"), "junkjunk") self.assertEqual(svcrsc.xpath("GROUP/PARAM[@name='knurr']")[0] .get("datatype"), "char") self.assertEqual(len(svcrsc.xpath("PARAM[@name='standardID']")), 0) self.assertEqual(svcrsc.xpath("PARAM[@name='accessURL']")[0] .get("value"), "http://localhost:8080/temporary/dl2/dlget") def testExt2Rsc(self): svcrsc = self.treeAndRD[0].xpath("RESOURCE[@ID='ext_svc-02']")[0] self.assertEqual(svcrsc.get("name"), "dl3") self.assertEqual(svcrsc.xpath("PARAM[@name='standardID']")[0] .get("value"), "ivo://ivoa.net/std/soda#sync-1.0") self.assertEqual(svcrsc.xpath("PARAM[@name='accessURL']")[0] .get("value"), "http://localhost:8080/temporary/dl3/dlget") self.assertEqual(svcrsc.xpath("GROUP/PARAM[@name='knurr']")[0] .get("datatype"), "char") self.assertEqual(svcrsc.xpath("GROUP/PARAM[@name='ruck']")[0] .get("datatype"), "int") def testWrongProcLinkType(self): self.assertRaisesWithMsg(api.StructureError, "ProcLinkDef must be constructed with a dlget-renderable service.", datalink.ProcLinkDef, ('ivo://x-gone', api.resolveCrossId("//adql#query"))) if __name__=="__main__": testhelpers.main(DatalinkMetaRowsTest)