"""
Tests for event propagation and user interaction.
"""

#c Copyright 2008-2019, 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 contextlib
import datetime
import os
import re
import shutil
import subprocess
import sys
import traceback

from gavo.helpers import testhelpers

from gavo import base
from gavo import api
from gavo import rsc
from gavo import rscdesc
from gavo import utils
from gavo.base import events
from gavo.helpers import testtricks
from gavo.user import cli

import tresc


class MiscCLITest(testhelpers.VerboseTest):
	resources = [("conn", tresc.dbConnection)]

	def testUnhandledException(self):
		self.assertOutput(cli.main, argList=["raise"], 
			expectedStderr=lambda msg: "Unhandled exception" in msg,
			expectedRetcode=1)

	def testTopLevelHelp(self):
		self.assertOutput(cli.main, argList=["--help"], 
			expectedStdout=lambda msg: "<func> is a unique" in msg)

	def testNonUniqueMatch(self):
		self.assertOutput(cli.main, argList=["a"], 
			expectedRetcode=1,
			expectedStderr=
				lambda msg: msg.startswith("Multiple matches for function a"))

	def testNoMatch(self):
		self.assertOutput(cli.main, argList=["xyloph"], 
			expectedRetcode=1,
			expectedStderr=
				lambda msg: msg.startswith("No match for function xyloph"))

	def testSubCmdHelp(self):
		self.assertOutput(cli.main, argList=["publish", "--help"], 
			expectedStdout=lambda msg: 
				"usage: dachs publish [-h] [-a] [-m] [-k] [-u] rd [rd ...]" in msg)

	def testAProg(self):
		self.assertOutput(cli.main, argList=["stc", "utypes", "Position ICRS"],
			expectedStdout=lambda msg: 
				"AstroCoordSystem.SpaceFrame.CoordRefFrame" in msg)

	def testCleanTAP(self):
		self.conn.commit()
		self.assertOutput(cli.main, argList=["admin", "cleantap"],
			expectedRetcode=0)

	def testExecute(self):
		self.assertOutput(cli.main, argList=["admin", "exec", 
			"data/regtest#silly"],
			expectedStdout="Spawning thread for cron job Do silly things\nSilly\n")

	def testCleanTAPPending(self):
		self.conn.commit()
		self.assertOutput(cli.main, argList=["admin", "cleantap", "-p"],
			expectedRetcode=0, expectedStderr="")

	def testShowDD(self):
		self.assertOutput(cli.main, argList=["show", "dds", "//tap"],
			expectedStdout="importTablesFromRD\n"
				"importDMsFromRD\nimportColumnsFromRD\nimportGroupsFromRD\n"
				"importFkeysFromRD\ncreateSchema*\ncreateJobTable*\n")

	def testShowDDBad(self):
		self.assertOutput(cli.main, argList=["show", "dds", "//nonex"],
			expectedRetcode=1,
			expectedStderr=lambda msg: 
				"resource descriptor '__system__/nonex' could not be located")

	def testVersion(self):
		self.assertOutput(cli.main, argList=["--version"],
			expectedRetcode=0,
			expectedStdout=lambda msg: re.match(
				r"Software \(\d+\.\d+(\.\d+)?\)"
				r" Schema \(\d+/-?\d+\)" "\n", msg))

	def testDachsInitJustRuns(self):
		from gavo.user import initdachs

		self.assertOutput(initdachs.main, argList=[],
			expectedRetcode=0,
			expectedStdout="",
			expectedStderr="")


class _CatalogTestResource(tresc.FileResource):
	path = "inputs/somecat/q.rd"
	content = """<resource schema="somecat">
			<meta name="creationDate">2000-01-01T00:00:00</meta>
			<meta name="title">A test resource</meta>
			<meta name="description">(for DaCHS)</meta>
			<meta name="subject">Objectivity</meta>
			<table onDisk="True" id="cat">
				<column name="ra" ucd="pos.eq.ra;meta.main"/>
				<column name="dec" ucd="pos.eq.dec;meta.main"/>
				<column name="t_min" type="timestamp"/>
				<column name="t_max" type="timestamp"/>
				<publish/>
			</table>
			<coverage>
				<updater sourceTable="cat"/>
				<spatial>0/0-11</spatial>
				<temporal>0 6e4</temporal>
			</coverage>
			<data id="import">
				<sources><item>1</item><item>4</item><item>x</item></sources>
				<embeddedGrammar>
					<iterator><code>
						seed = float(self.sourceToken)
						yield {
							"ra": seed+4,
							"dec": 2-seed,
							"t_min": datetime.datetime(int(2010+seed), int(seed)%12, 1, 10),
							"t_max": datetime.datetime(int(2010+seed), int(seed)%12, 1, 12),
						}
					</code></iterator>
				</embeddedGrammar>
				<make table="cat"/>
			</data>
		</resource>"""


class ImportTest(testhelpers.VerboseTest):
	resources = [("conn", tresc.dbConnection),
		("res", _CatalogTestResource())]

	def testLifecycle(self):
		self.conn.commit()
		with base.AdhocQuerier() as querier:
			self.assertOutput(cli.main, 
				argList=["--suppress-log",
					"imp", "-c", "somecat/q"],
				stdoutStrings=["Rows affected: 2"],
				expectedRetcode=101)

			self.failIf(querier.getTableType("somecat.cat") is None)
			self.failIf(querier.getTableType("test.typesTable") is not None)

			self.assertOutput(cli.main,
				argList=["--disable-spew", "--suppress-log", "publish", "somecat/q"],
				expectedRetcode=0, expectedStdout="")
			
			self.failUnless(list(querier.query("SELECT * FROM dc.subjects"
				" WHERE subject=%(s)s", {'s': "Objectivity"})))

			self.assertOutput(cli.main,
				argList=["--disable-spew", "--suppress-log", "limits", "somecat/q"],
				expectedRetcode=0, expectedStdout="")

			changedRd = base.caches.getRD("somecat/q")
			self.assertEqual(changedRd.coverage.spatial, "6/18064,18098")
			self.assertEqual(changedRd.coverage.spectral, [])
			self.assertEqual(changedRd.coverage.temporal, [(55562.4, 56748.5)])

			# drop it all, make sure all traces are gone
			self.assertOutput(cli.main,
				argList=["--disable-spew", "--suppress-log",
					"drop", "somecat/q"], expectedStdout="", expectedStderr="")
			self.failIf(list(querier.query("SELECT * FROM dc.subjects"
				" WHERE subject=%(s)s", {'s': "Objectivity"})))
			self.failUnless(querier.getTableType("comecat.cat") is None)

	def testImportDeniedForOffInputs(self):
		destName = os.path.expanduser("~/foobar.rd")
		with testhelpers.testFile(destName, '<resource schema="junk"/>'):
			self.assertOutput(cli.main, 
				argList=["imp", destName],
				expectedRetcode=1, expectedStderr=lambda tx:
				re.match(r"\*\*\* Error: .*/foobar.rd: Only RDs below inputsDir"
				"\n.*/inputs. are allowed.\n", tx) is not None)

	def testMetaImportAndPurge(self):
		self.assertOutput(cli.main, argList=["purge", "test.adql"])
		# make sure the test schema exists before running the test
		self.conn.commit()
		self.assertOutput(cli.main, 
			argList=["imp", "data/test", "productimport-skip"])
		try:
			with base.AdhocQuerier(base.getWritableAdminConn) as querier:
				querier.query("CREATE TABLE test.adql (erratic INTEGER)")
				querier.query("INSERT INTO test.adql VALUES (1)")

			with base.AdhocQuerier() as querier:
				self.assertOutput(cli.main, argList=
					["imp", "-m", "data/test", "ADQLTest"],
					expectedStdout=
						lambda output: "Updating meta for ADQLTest\n" in output)
				self.assertEqual(list(querier.query(
					"select tablename, sourcerd, adql"
					"  from dc.tablemeta where tablename='test.adql'")),
					[(u'test.adql', u'data/test', True)])
				
				# make sure gavo imp didn't touch the table
				self.assertEqual(list(querier.query("SELECT * FROM test.adql")),
					[(1,)])
		finally:
			self.assertOutput(cli.main, argList=["purge", "test.adql"])

	def testNonExistingDataImpError(self):
		with testtricks.testFile(
				os.path.join(base.getConfig("inputsDir"), "empty.rd"),
				"""<resource schema="test"><table id="foo"/></resource>"""):
			self.assertOutput(cli.main, argList=["--hints", "imp", "empty", "foo"],
				expectedRetcode=1, expectedStderr="*** Error: The DD 'foo'"
					" you are trying to import is not defined within\nthe RD"
					" 'empty'.\nHint: Data elements available in empty include (None)\n")

	def testNoAutoDataImpError(self):
		with testtricks.testFile(
				os.path.join(base.getConfig("inputsDir"), "empty.rd"),
				"""<resource schema="test"><table id="foo"/>
				<data auto="False" id="x"><make table="foo"/></data>
				<data auto="False" id="y"><make table="foo"/></data>
				</resource>"""):
			self.assertOutput(cli.main, argList=["--hints", "imp", "empty"],
				expectedRetcode=1, expectedStderr='*** Error: Neither automatic'
					' not manual data selected from RD empty\nHint: All data'
					' elements have auto=False.  You have to explicitely name\none'
					' or more data to import (names available: x, y)\n')

	def testMetaNotPropagatedMessage(self):
		self.assertOutput(cli.main, 
			argList=["imp", "-m", "data/test", "recaftertest"],
			expectedRetcode=0,
			expectedStdout=lambda msg: ("not rebuilding dependencies." in msg)
				and (": data/test#import_pythonscript" in msg))

	def testPubUnpub(self):
		self.assertOutput(cli.main,
			argList=["--disable-spew", "--suppress-log", "publish", "somecat/q"],
			expectedRetcode=0, expectedStdout="")
		self.assertEqual(
			list(self.conn.query("SELECT resid, deleted FROM dc.resources"
			" WHERE sourcerd='somecat/q'")),
			[(u'cat', False)])
		self.assertOutput(cli.main,
			argList=["--disable-spew", "--suppress-log", "publish", "-u", 
				"somecat/q"], expectedRetcode=0, expectedStdout="")
		self.assertEqual(
			list(self.conn.query("SELECT resid, deleted FROM dc.resources"
			" WHERE sourcerd='somecat/q'")),
			[(u'cat', True)])


class _SysRDResource(tresc.FileResource):
	path = "inputs/sysrd.rd"
	content = """<resource schema="test">
			<table onDisk="True" id="fromclitest" system="True">
				<column name="x" type="integer"><values nullLiteral="0"/></column>
			</table>
			<data id="y"><make table="fromclitest"/></data>
		</resource>"""


class SystemImportTest(testhelpers.VerboseTest):
	resources = [("conn", tresc.dbConnection),
		("sysrdFile", _SysRDResource())]

	def _fillTable(self):
		rd = api.getRD("sysrd")
		t = rsc.TableForDef(rd.getById("fromclitest"), connection=self.conn)
		t.addRow({'x': 2})
		self.conn.commit()

	def testNoSystemImportDefault(self):
		self.assertOutput(cli.main, argList=["imp", "--system", "sysrd"],
			expectedRetcode=0, expectedStderr="")
		# Write a 2 into the table that must survive the next imp
		self._fillTable()

		self.assertOutput(cli.main, argList=["imp", "sysrd"],
			expectedRetcode=0, expectedStderr="")
		with base.AdhocQuerier() as q:
			self.assertEqual(list(q.query("select * from test.fromclitest")),
				[(2,)])
			
	def testSystemImport(self):
		self.assertOutput(cli.main, argList=["imp", "--system", "sysrd"],
			expectedRetcode=0, expectedStderr="")
		# Write a 2 into the table that must survive the next imp
		self._fillTable()

		self.assertOutput(cli.main, argList=["imp", "--system", "sysrd"],
			expectedRetcode=0, expectedStderr="")
		with base.AdhocQuerier() as q:
			self.assertEqual(list(q.query("select * from test.fromclitest")),
				[])

	def testSystemDropDrops(self):
		self.assertOutput(cli.main, argList=["drop", "--system", "sysrd"],
			expectedRetcode=0, expectedStderr="", expectedStdout="")
		with base.AdhocQuerier() as q:
			self.failUnless(q.getTableType("test.fromclitest") is None)


class _FITSGeneratedRD(testhelpers.TestResource):
	def make(self, ignored):
		p = testhelpers.ForkingSubprocess(
			["test harness", "--debug", "mkrd", "-r", 
				str(os.path.join(base.getConfig("inputsDir"), "data")),
				"test_data/ex.fits"],
			executable=cli.main, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
		out, err = p.communicate(input=input)
		retcode = p.wait()
		if err or retcode:
			sys.stderr.write("panic: generating RD failed, bailing out.\n%s\n"%
				err)
			sys.exit(0)
		return out


class MkRDTest(testhelpers.VerboseTest, testhelpers.XSDTestMixin):
	resources = [("fitsrd", _FITSGeneratedRD()), ('oc', tresc.obscoreTable)]

	def testFITSRDLooksOk(self):
		for frag in [
				'<column name="color" type="text"',
				'description="Effective exposure time [seconds]"',
				'<map key="exptime">EXPTIME</map>']:
			self.failUnless(frag in self.fitsrd, "%s missing"%frag)
	
	def testRunImp(self):
		with testhelpers.testFile(
				os.path.join(base.getConfig("inputsDir"), "gen.rd"), 
				self.fitsrd.original):
			self.assertOutput(cli.main, ["imp", "gen"], expectedStderr="",
				expectedStdout=lambda s: "Rows affected: 1", expectedRetcode=0)
			self.assertOutput(cli.main, ["drop", "gen"])

	def testFITSRDMetaFirst(self):
		self.failUnless(self.fitsrd.split("\n")[1].strip().startswith("<meta"))

	def testStartList(self):
		self.assertOutput(cli.main, ["start", "list"],
			expectedStdout=lambda s: "scs -- Catalogs via SCS and TAP" in s)

	def testStartBadCall(self):
		self.assertOutput(cli.main, ["start", "knakurtt"],
			expectedRetcode=1,
			expectedStdout="*X*X* No template for knakurtt.\n")

	def testStartSCS(self):
		def makeAsserter(resdir):
			def assertThings(output):
				self.assertEqual(output, "")
				with open("q.rd") as f:
					generated = f.read()

				self.assertWellformed(generated)

				self.failUnless(
					'schema="%s"'%resdir in generated)
				self.failUnless(
					"%s-"%datetime.datetime.utcnow().year
					in generated)
				self.failUnless(
					'<meta name="creator">%authors' in generated)

				return True
			return assertThings

		with utils.sandbox(tmpdir=os.path.join(base.getConfig("inputsDir"))):
			resdir = os.path.split(os.getcwd())[-1]
			self.assertOutput(cli.main, ["start", "scs"],
				expectedStdout=makeAsserter(resdir))

	def testStartSIAP(self):
		def makeAsserter(resdir):
			def assertThings(output):
				self.assertEqual(output, "")
				with open("q.rd") as f:
					generated = f.read()

				self.failUnless(
					'schema="%s"'%resdir in generated)
				self.assertWellformed(generated)
				self.failUnless('//obscore#publishSIAP' in generated)
				return True

			return assertThings

		with utils.sandbox(tmpdir=os.path.join(base.getConfig("inputsDir"))):
			resdir = os.path.split(os.getcwd())[-1]
			self.assertOutput(cli.main, ["start", "siap"],
				expectedStdout=makeAsserter(resdir))

	def testStartSSAP(self):
		def makeAsserter(resdir):
			def assertThings(output):
				self.assertEqual(output, "")
				with open("q.rd") as f:
					generated = f.read()

				self.failUnless(
					'schema="%s"'%resdir in generated)
				self.assertWellformed(generated)
				self.failUnless('//obscore#publishSSAPMIXC' in generated)
				return True

			return assertThings

		with utils.sandbox(tmpdir=os.path.join(base.getConfig("inputsDir"))):
			resdir = os.path.split(os.getcwd())[-1]
			self.assertOutput(cli.main, ["start", "ssap+datalink"],
				expectedStdout=makeAsserter(resdir))

	def testStartEPNTAP(self):
		def makeAsserter(resdir):
			def assertThings(output):
				self.assertEqual(output, "")
				with open("q.rd") as f:
					generated = f.read()

				self.failUnless(
					'schema="%s"'%resdir in generated)
				self.assertWellformed(generated)
				self.failUnless('>//epntap2#localfile-2_0' in generated)
				return True

			return assertThings

		with utils.sandbox(tmpdir=os.path.join(base.getConfig("inputsDir"))):
			resdir = os.path.split(os.getcwd())[-1]
			self.assertOutput(cli.main, ["start", "epntap"],
				expectedStdout=makeAsserter(resdir))

class _MyRDResource(tresc.FileResource):
	path = "inputs/myrd.rd"
	content = """<resource schema="test">
			<table onDisk="True" id="autotable">
				<column name="x" type="integer" required="True"/></table>
			<table onDisk="True" id="noautotable">
				<column name="x" type="integer" required="True"/></table>
			<data id="y"><make table="autotable"/></data>
			<data id="z" auto="False"><make table="noautotable"/></data>
		</resource>"""


class DropTest(testhelpers.VerboseTest):
	resources = [("myrdFile", _MyRDResource())]

	def testAutoDropping(self):
		self.assertOutput(cli.main, argList=["imp", "myrd", "y", "z"],
			expectedRetcode=0, expectedStderr="")
		self.assertOutput(cli.main, argList=["drop", "myrd"],
			expectedRetcode=0, expectedStderr="")
		with base.AdhocQuerier() as q:
			self.failUnless(q.getTableType("test.autotable") is None)
			self.failIf(q.getTableType("test.noautotable") is None)

	def testAllDropping(self):
		self.assertOutput(cli.main, argList=["imp", "myrd", "y", "z"],
			expectedRetcode=0, expectedStderr="")
		self.assertOutput(cli.main, argList=["drop", "myrd", "--all"],
			expectedRetcode=0, expectedStderr="")
		with base.AdhocQuerier() as q:
			self.failUnless(q.getTableType("test.autotable") is None)
			self.failUnless(q.getTableType("test.noautotable") is None)

	def testNamedDropping(self):
		self.assertOutput(cli.main, argList=["imp", "myrd", "y", "z"],
			expectedRetcode=0, expectedStderr="")
		self.assertOutput(cli.main, argList=["drop", "myrd", "z"],
			expectedRetcode=0, expectedStderr="")
		with base.AdhocQuerier() as q:
			self.failIf(q.getTableType("test.autotable") is None)
			self.failUnless(q.getTableType("test.noautotable") is None)


class ValidationTest(testhelpers.VerboseTest):
	def testValidUserconfig(self):
		base.caches.clearForName(rscdesc.USERCONFIG_RD_PATH)
		with testtricks.testFile(rscdesc.USERCONFIG_RD_PATH+".rd",
				"""<resource schema="test"><STREAM id="foo"><column name="abc"/>
				</STREAM></resource>"""):
			self.assertOutput(cli.main, argList=["val", "%"],
				expectedRetcode=0, expectedStderr='',
				expectedStdout='% -- OK\n')

	def testInvalidUserconfig(self):
		base.caches.clearForName(rscdesc.USERCONFIG_RD_PATH)
		with testtricks.testFile(rscdesc.USERCONFIG_RD_PATH+".rd",
				"""<resource schema="test"><STREAM id="foo"><column name="abc">
				</STREAM></resource>"""):
			self.assertOutput(cli.main, argList=["val", "%"],
				expectedRetcode=0, expectedStderr='',
				expectedStdout=lambda text: re.match(
					r'% -- \[ERROR\] %: Malformed RD input, message follows\s+'
					r"\*\*\* Error: .*etc/userconfig.rd"
					"\s+mismatched\s+tag:\s+line\s+2,\s+column 6", text))

	def testMiscValidation(self):
		with testhelpers.testFile(name=None, inDir=base.getConfig("inputsDir"),
				content="""<resource schema="notexistingXXX">
					<table id="t1"/>
					<table id="t1"/>
					</resource>""") as path:
			rdId = path.split("/")[-1]

			def doAssertions(output):
				self.assertTrue("[WARNING] %s: RD %s: resource directory"
					" '%s/inputs/notexistingXXX' does not exist"%(
					rdId, rdId, base.getConfig("rootDir"))
					in output)

				self.assertTrue("%s -- [WARNING] %s: Element with id t1 overwritten."%(
					rdId, rdId))

				self.assertTrue(output.endswith("OK\n"))
				return True

			self.assertOutput(cli.main, argList=["val", rdId],
				expectedRetcode=0, expectedStderr='',
				expectedStdout=doAssertions)
		

class CLIReferenceTest(testhelpers.VerboseTest):
	def testNonExRD(self):
		self.assertRaisesWithMsg(base.RDNotFound,
			"Resource descriptor 'i/do/not/exist' could not be located"
			" in file system",
			api.getReferencedElement,
			("i/do/not/exist",))
	
	def testNoRDReference(self):
		with testhelpers.testFile(os.path.join(base.getConfig("inputsDir"),
				"nordref.rd"),
				"""<resource schema="__test"><table original="i/do/not#exist"/>
				</resource>""") as src:
			self.assertRaisesWithMsg(base.RDNotFound,
				"Resource descriptor u'i/do/not' could not be located in file system",
				api.getReferencedElement,
				("nordref",))

	def testNoElDirectReference(self):
		self.assertRaisesWithMsg(base.NotFoundError,
			"Element with id 'justrandomjunk' could not be located in RD data/test",
			api.getReferencedElement,
			("data/test#justrandomjunk",))

	def testNoElIndirectReference(self):
		with testhelpers.testFile(os.path.join(base.getConfig("inputsDir"),
				"badref.rd"),
				"""<resource schema="__test">
					<table original="data/test#justrandomjunk"/>
				</resource>""") as src:
			self.assertRaisesWithMsg(base.NotFoundError,
				"Element with id u'justrandomjunk' could not be located"
				" in RD data/test",
				api.getReferencedElement,
				("badref",))

	def testNoElIndirectReferenceInDir(self):
		baseDir = os.path.join(base.getConfig("inputsDir"), "test")
		with testhelpers.testFile(os.path.join(baseDir, "locbad.rd"),
				"""<resource schema="__test">
					<table original="data/test#justrandomjunk"/>
				</resource>""") as src:
			with utils.in_dir(baseDir):
				self.assertRaisesWithMsg(base.NotFoundError,
					"Element with id u'justrandomjunk' could not be located"
					" in RD data/test",
					api.getReferencedElement,
					("locbad",))


class DumpingTest(testhelpers.VerboseTest):
	resources = [("table", tresc.csTestTable)]

	def _saveOutput(self, owd):
		for fName in ["output.stderr", "output.stdout"]:
			if os.path.exists(fName):
				shutil.move(fName, os.path.join(owd, fName))

	def testRestoreMultiOverwriting(self):
		nrowsBefore = list(self.table.connection.query(
			"select count(*) from tap_schema.tables"))
		self.table.connection.commit()

		with utils.sandbox(extractfunc=self._saveOutput):
			self.assertOutput(cli.main, 
				argList=["dump", "create", "tmp.ddump", 
					self.table.tableDef.getFullId(),
					"//tap#tables"],
				expectedRetcode=0,
				expectedStdout="")
			self.assertOutput(cli.main, 
				argList=["dump", "load", "tmp.ddump"],
				expectedRetcode=0,
				expectedStdout="")
		
		nrows = list(self.table.connection.query("select count(*) from %s"%
			self.table.tableDef.getQName()))[0][0]
		self.assertEqual(nrows, 1)
		nrows = list(self.table.connection.query(
			"select count(*) from tap_schema.tables"))
		self.assertEqual(nrows, nrowsBefore)

	def testRestoreDeleted(self):
		tableId = self.table.tableDef.getFullId()

		with utils.sandbox(extractfunc=self._saveOutput):
			self.assertOutput(cli.main, 
				argList=["dump", "create", "tmp.dump", tableId],
				expectedRetcode=0,
				expectedStdout="")

			self.assertOutput(cli.main, 
				argList=["purge", self.table.tableDef.getQName()])
			self.assertEqual(self.table.exists(), False)

			self.assertOutput(cli.main, 
				argList=["dump", "load", "tmp.dump"],
				expectedRetcode=0,
				expectedStdout="")
		
		nrows = list(self.table.connection.query("select count(*) from %s"%
			self.table.tableDef.getQName()))[0][0]
		self.assertEqual(nrows, 1)

	def testLsDump(self):
		def assertTablesListed(output):
			self.assertTrue("data/test#csdata" in output)
			self.assertTrue("__system__/tap#supportedmodels" in output)
			self.assertTrue(" probably " in output)
			self.assertTrue(" 59" in output)  #len(csdata)
			return True

		with utils.sandbox(extractfunc=self._saveOutput):
			self.assertOutput(cli.main, 
				argList=["dump", "create", "tmp.ddump", 
					self.table.tableDef.getFullId(),
					"//tap"],
				expectedRetcode=0,
				expectedStdout="")
			self.assertOutput(cli.main, 
				argList=["dump", "ls", "tmp.ddump"],
				expectedRetcode=0,
				expectedStdout=assertTablesListed)


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