"""
Tests for the products infrastructure.
"""

#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 io
import os
import tarfile

from gavo.helpers import testhelpers

from gavo import api
from gavo import base
from gavo import formats
from gavo import rscdef
from gavo import svcs
from gavo import votable
from gavo.helpers import testtricks
from gavo.helpers import trialhelpers
from gavo.protocols import products
from gavo.web import producttar

import tresc


class StandardPubDIDTest(testhelpers.VerboseTest):
	def testMakeSPD(self):
		self.assertEqual(rscdef.getStandardPubDID("a/b/c"),
			"ivo://x-testing/~?a/b/c")
	
	def testParseSPD(self):
		self.assertEqual(
			rscdef.getAccrefFromStandardPubDID("ivo://x-testing/~?a/b/c"),
			"a/b/c")
	
	def testRejectParseSPD(self):
		self.assertRaisesWithMsg(base.NotFoundError,
			"The authority in the dataset identifier 'ivo://quatsch/batsch'"
			" could not be located in the authorities managed here",
			rscdef.getAccrefFromStandardPubDID,
			("ivo://quatsch/batsch",))


class _TestWithProductsTable(testhelpers.VerboseTest):
	resources = [('conn', tresc.prodtestTable), ('users', tresc.testUsers)]

	def setUp(self):
		testhelpers.VerboseTest.setUp(self)
		self.service = api.getRD("//products").getById("p")


class TarTest(_TestWithProductsTable):
	def setUp(self):
		_TestWithProductsTable.setUp(self)
		self.tarService = self.service.rd.getById("getTar")

	def _getTar(self, inDict, qm=None):
		if qm is None:
			qm = svcs.QueryMeta()
		res = self.tarService.run("form", inDict, qm)
		dest = io.BytesIO()
		producttar.getTarMaker()._productsToTar(res, dest)
		return dest.getvalue()

	def _assertIsTar(self, res):
		f = tarfile.open("data.tar", "r:*", io.BytesIO(res))
		f.close()

	def testFreeNoAuth(self):
		res = self._getTar({"pattern": "test.prodtest#data/b.imp"})
		self._assertIsTar(res)
		self.assertTrue(b"\nobject: michael" in res)

	def testAllNoAuth(self):
		res = self._getTar({"pattern": "test.prodtest#%"})
		self._assertIsTar(res)
		self.assertTrue(b"\nobject: michael" in res)
		self.assertTrue(b"This file is embargoed.  Sorry" in res)
		self.assertFalse(b"\nobject: gabriel" in res)
	
	def testAllWithAuth(self):
		qm = svcs.QueryMeta.fromRequest(
			trialhelpers.FakeRequest(user="X_test", password="megapass"))
		res = self._getTar({"pattern": "test.prodtest#%"}, qm)
		self._assertIsTar(res)
		self.assertTrue(b"\nobject: michael" in res)
		self.assertFalse(b"This file is embargoed.  Sorry" in res)
		self.assertTrue(b"\nobject: gabriel" in res)
	
	def testAllWithWrongAuth(self):
		qm = svcs.QueryMeta()
		qm["user"], qm["password"] = "Y_test", "megapass"
		res = self._getTar({"pattern": "test.prodtest#%"}, qm)
		self._assertIsTar(res)
		self.assertTrue(b"\nobject: michael" in res)
		self.assertTrue(b"This file is embargoed.  Sorry" in res)
		self.assertFalse(b"\nobject: gabriel" in res)

	def testAccrefs(self):
		res = self._getTar({"accref": ["data/a.imp", "data/b.imp",
			"data/is-missing.dat"]})
		tf = tarfile.open("data.tar", "r:*", io.BytesIO(res))
		# TODO: do we really want to silently ignore missing files?
		self.assertEqual(
			[member.name for member in tf],
			['dc_data/a.imp', 'dc_data/b.imp'])


def _FakeRequest(**kwargs):
	return trialhelpers.FakeRequest("/",
		args=trialhelpers.makeRequestArgs(**kwargs))


class RaccrefTest(_TestWithProductsTable):

	# tests for dcc: with SDM VOTables are in ssatest.py

	def	testBadConstructurVals(self):
		self.assertRaisesWithMsg(base.ValidationError,
			"Field accref: Invalid value for constructor argument to RAccref:"
				" scale='klonk'",
			products.RAccref,
			("testing", {"scale": "klonk"}))

	def testExtraParamsIgnored(self):
		pk = products.RAccref("name", {"sra": "3", "ignored": True})
		self.assertEqual(pk.accref, "name")
		self.assertEqual(pk.params, {"sra": 3.})

	def testSerialization(self):
		pk = products.RAccref(
			"extra weird/product+name%something.fits",
			{"scale": "4"})
		self.assertEqual(str(pk),
			"extra%20weird/product%2Bname%25something.fits?scale=4")

	def testFromRequestSimple(self):
		pk = products.RAccref.fromRequest("extra weird+key",
			_FakeRequest())
		self.assertEqual(pk.accref, "extra weird+key")
		self.assertEqual(pk.params, {})

	def testFromStringWithArgs(self):
		pk = products.RAccref.fromString(
			"extra%20weird&?%2bkey?ra=2&sra=0.5&dec=4&sdec=0.75")
		self.assertEqual(pk.accref, "extra weird&?+key")
		self.assertEqual(pk.params, {"ra": 2, "sra":0.5, "dec":4, "sdec":0.75})

	def testFromStringWithoutArgs(self):
		pk = products.RAccref.fromString("extra%20weird&%2bkey")
		self.assertEqual(pk.accref, "extra weird&+key")
		self.assertEqual(pk.params, {})
	
	def testBadFromString(self):
		self.assertRaisesWithMsg(base.ValidationError,
			"Field accref: Invalid value for constructor argument to RAccref:"
			" sra='huhu'",
			products.RAccref.fromString,
			("worz?sra=huhu",))

	def testProductsRowRaises(self):
		nonExProd = products.RAccref("junkomatix/@@ridiculosa")
		self.assertRaisesWithMsg(base.NotFoundError,
			"accref 'junkomatix/@@ridiculosa' could not be located in product table",
			lambda: nonExProd.productsRow,
			())

	def testProductsRowReturns(self):
		prod = products.RAccref("data/a.imp")
		self.assertEqual(prod.productsRow, {
			'embargo': datetime.date(2030, 12, 31),
			'accessPath': 'data/a.imp',
			'mime': 'text/plain',
			'owner': 'X_test',
			'accref': 'data/a.imp',
			'datalink': None,
			'preview': 'data/broken.imp',
			'preview_mime': "text/plain",
			'sourceTable': 'test.prodtest'})

	def testPreview(self):
		prod = products.RAccref.fromString("data/a.imp?preview=true")
		self.assertEqual(prod.params, {"preview": True})


class ProductsCoreTest(_TestWithProductsTable):
	def _getProductFor(self, accref, moreFields={}):
		inData = {"accref": [products.RAccref.fromString(accref)]}
		inData.update(moreFields)
		svc = base.caches.getRD("//products").getById("p")
		rows = trialhelpers.runSvcWith(svc, "get", inData)
		return testhelpers.pickSingle(rows)

	def _getOutput(self, prod):
		return b"".join(prod.iterData())

	def testBasic(self):
		res = self._getProductFor("data/b.imp")
		self.assertTrue(isinstance(res, products.FileProduct))
		self.assertTrue(self._getOutput(res).startswith(
			b"alpha: 03 34 33.45"))

	def testNonExistingProduct(self):
		res = self._getProductFor("junk/kotter")
		self.assertTrue(isinstance(res, products.NonExistingProduct))
		self.assertRaisesWithMsg(IOError,
			"junk/kotter does not exist",
			self._getOutput,
			(res,))
		self.assertRaisesWithMsg(svcs.UnknownURI,
			"No dataset with accref junk/kotter known here.",
			res.render,
			(None,))
	
	def testRemovedProduct(self):
		srcPath = os.path.join(base.getConfig("inputsDir"), "data", "b.imp")
		os.rename(srcPath, srcPath+".bak")
		try:
			res = self._getProductFor("data/b.imp")
			self.assertTrue(isinstance(res, products.InvalidProduct))
		finally:
			os.rename(srcPath+".bak", srcPath)

	def testProtectedProductUnauth(self):
		res = self._getProductFor("data/a.imp")
		self.assertTrue(isinstance(res, products.UnauthorizedProduct))

	def testProtectedProductWithMoreArg(self):
		res = self._getProductFor("data/a.imp?scale=2")
		self.assertTrue(isinstance(res, products.UnauthorizedProduct))

	def testProtectedProductBadAuth(self):
		res = self._getProductFor("data/a.imp",
			{"user": "Y_test", "password": "megapass"})
		self.assertTrue(isinstance(res, products.UnauthorizedProduct))

	def testProtectedAuth(self):
		res = self._getProductFor("data/a.imp",
			{"user": "X_test", "password": "megapass"})
		self.assertTrue(isinstance(res, products.FileProduct))
		self.assertTrue(self._getOutput(res).startswith(
			b"alpha: 23 34 33.45"))

	def testRemoteProduct(self):
		with tresc.prodtestTable.prodtblRow(accessPath="http://foo.bar"):
			res = self._getProductFor("just.testing/nowhere")
			self.assertTrue(isinstance(res, products.RemoteProduct))
			self.assertRaisesWithMsg(svcs.WebRedirect,
				"This is supposed to redirect to http://foo.bar",
				res.render,
				(None,))

	def testInvalidProduct(self):
		with tresc.prodtestTable.prodtblRow(accessPath="/non/existing/file"):
			res = self._getProductFor("just.testing/nowhere")
			self.assertTrue(isinstance(res, products.InvalidProduct))
			self.assertRaises(svcs.UnknownURI,
				res.render,
				None)

# TODO: have test scaffolding (ex.fits?) to actually make these work
	def testScaledProduct(self):
		prod = self._getProductFor("data/b.imp?scale=3")
		self.assertEqual(str(prod), "<Invalid product data/b.imp?scale=3>")
	
	def testCutoutProduct(self):
		res = self._getProductFor("data/b.imp?ra=3&dec=4&sra=2&sdec=4")
		self.assertEqual(str(res), "<Invalid product data/b.imp?"
			"dec=4.0&ra=3.0&sdec=4.0&sra=2.0>")


class _FakeProduct(products.ProductBase):
	def iterData(self):
		yield b"1234"
		yield b"1234"
		yield b"    "*10
		yield b"end"


class FileIntfTest(ProductsCoreTest):
	def testFallbackBuffering(self):
		p = _FakeProduct(products.RAccref.fromString("data/a.imp"))
		self.assertEqual(p.read(1), b"1")
		self.assertEqual(p.read(1), b"2")
		self.assertEqual(p.read(7), b"341234 ")
		rest = p.read()
		self.assertEqual(len(rest), 42)
		self.assertEqual(rest[-4:], b" end")
		p.close()
	
	def testNativeRead(self):
		p = self._getProductFor("data/a.imp")
		self.assertEqual(p.read(10), b"alpha: 23 ")
		self.assertTrue(hasattr(p._openedInputFile, "read"))
		p.close()
		self.assertEqual(p._openedInputFile, None)


class StaticPreviewTest(testhelpers.VerboseTest):

	resources = [('conn', tresc.prodtestTable), ('users', tresc.testUsers)]

	def testStaticPreviewLocal(self):
		prod = products.getProductForRAccref("data/a.imp?preview=True")
		self.assertTrue(isinstance(prod, products.StaticPreview))
		self.assertEqual(prod.read(200), b'kaputt.\n')

	def testStaticPreviewRemote(self):
		prod = products.getProductForRAccref("data/b.imp?preview=True")
		self.assertTrue(isinstance(prod, products.RemotePreview))
		self.assertEqual(str(prod),
			'<Remote image/jpeg at http://example.com/borken.jpg>')


class AutoPreviewTest(testhelpers.VerboseTest):

	resources = [('fits', tresc.fitsTable)]

	def testAutoPreviewMiss(self):
		prod = products.getProductForRAccref("data/ex.fits?preview=True")
		self.assertTrue(isinstance(prod, products.FileProduct))

	def testAutoPreviewHit(self):
		cacheLocation = products.PreviewCacheManager.getCacheName(
			"data/ex.fits")
		with testtricks.testFile(os.path.basename(cacheLocation),
				"Abc, die Katze", inDir=os.path.dirname(cacheLocation)):
			prod = products.getProductForRAccref("data/ex.fits?preview=True")
			self.assertTrue(isinstance(prod, products.StaticPreview))
			self.assertEqual(prod.read(200), b"Abc, die Katze")


class _GlobalFITSLinks(testhelpers.TestResource):
	resources = [("fitsTable", tresc.fitsTable)]

	def make(self, deps):
		svc = api.getRD("//products").getById("dl")
		data, metadata = votable.loads(trialhelpers.runSvcWith(svc, "dlmeta", {
			"ID": [rscdef.getStandardPubDID("data/excube.fits")]})[1])
		res = {}
		for link in metadata.iterDicts(data):
			res.setdefault(link["semantics"], []).append(link)
		return res


class GlobalDatalinkTest(testhelpers.VerboseTest):
	resources = [("links", _GlobalFITSLinks())]

	def testNumberOfLinks(self):
		self.assertEqual(sum(len(r) for r in list(self.links.values())), 2)
	
	def testDatasetMeta(self):
		r, = self.links["#this"]
		self.assertEqual(tuple(r[s] for s in
			"content_length description error_message content_type ID".split()),
			(5760, "The full dataset.", None, "application/fits",
				"ivo://x-testing/~?data/excube.fits"))
	
	def testDatasetURL(self):
		self.assertEqual(self.links["#this"][0]["access_url"],
			"http://localhost:8080/getproduct/data/excube.fits")

	def testPreviewMeta(self):
		r, = self.links["#preview"]
		self.assertEqual(tuple(r[s] for s in
			"content_length description error_message content_type ID".split()),
			(None, "A preview for the dataset.", None, None,
				"ivo://x-testing/~?data/excube.fits"))
	
	def testPreviewURL(self):
		self.assertEqual(self.links["#preview"][0]["access_url"],
			"http://localhost:8080/getproduct/data/excube.fits?preview=True")


class MiscTest(testhelpers.VerboseTest):
	def testMediaGuessFITS(self):
		self.assertEqual(formats.guessMediaType("foo/bar/gaz42h+88.old.fits"),
			"application/fits")

	def testMediaGuessVOTable(self):
		self.assertEqual(formats.guessMediaType("bal.vot"),
			'application/x-votable+xml')

	def testFallback(self):
		self.assertEqual(formats.guessMediaType("voo/bar.d/basz"),
			"application/octet-stream")


class ProductLinkTest(testhelpers.VerboseTest):
	def testSimpleAccref(self):
		self.assertEqual(
			products.makeProductLink("foo/bar/baz.fits"),
			"http://localhost:8080/getproduct/foo/bar/baz.fits")

	def testWithoutHost(self):
		self.assertEqual(
			products.makeProductLink("foo/bar/baz.fits", withHost=False),
			"/getproduct/foo/bar/baz.fits")

	def testUseHost(self):
		self.assertEqual(
			products.makeProductLink("foo/bar/baz.fits",
				useHost="https://dc.g-vo.org"),
			"https://dc.g-vo.org/getproduct/foo/bar/baz.fits")

	def testURLIn(self):
		self.assertEqual(
			products.makeProductLink("http://example.org/foo/bar/baz.fits"),
			"http://example.org/foo/bar/baz.fits")
		self.assertEqual(
			products.makeProductLink("https://example.org/foo/bar/baz.fits"),
			"https://example.org/foo/bar/baz.fits")

	def testRAccrefIn(self):
		self.assertEqual(
			products.makeProductLink("http://example.org/foo/bar/baz.fits"
				"?ra=10&dec=20&sra=0.1&sdec=0.2&scale=2"),
			"http://example.org/foo/bar/baz.fits?ra=10&dec=20&sra=0.1"
				"&sdec=0.2&scale=2")


if __name__=="__main__":
	testhelpers.main(MiscTest)