src/test/org/apache/sqoop/metastore/SavedJobsTestBase.java - sqoop - Git at Google

 /**
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
  *
  *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */

 package org.apache.sqoop.metastore;

 import static org.apache.sqoop.metastore.GenericJobStorage.META_CONNECT_KEY;
 import static org.apache.sqoop.metastore.GenericJobStorage.META_PASSWORD_KEY;
 import static org.apache.sqoop.metastore.GenericJobStorage.META_USERNAME_KEY;

 import static org.hamcrest.core.IsCollectionContaining.hasItems;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertThat;

 import org.apache.sqoop.manager.ConnManager;
 import org.apache.sqoop.SqoopOptions;
 import org.apache.sqoop.tool.VersionTool;

 import org.apache.hadoop.conf.Configuration;
 import org.apache.sqoop.manager.DefaultManagerFactory;
 import org.apache.sqoop.tool.ImportTool;
 import org.apache.sqoop.metastore.JobData;
 import org.apache.sqoop.metastore.JobStorage;
 import org.apache.sqoop.metastore.JobStorageFactory;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.ExpectedException;

 import java.io.IOException;

 import java.sql.Connection;
 import java.sql.SQLException;
 import java.sql.Statement;

 import java.util.List;
 import java.util.Map;
 import java.util.TreeMap;

 /**
  * Test the metastore and job-handling features,
  * implemented for specific database services in sub-classes.
  */
 public abstract class SavedJobsTestBase {

   public static final String TEST_JOB = "testJob";
   public static final String TEST_TABLE_NAME = "abcd";
   public static final String TEST_TABLE_NAME_2 = "efgh";
   public static final String TEST_JOB_2 = "testJob2";
   public static final String TEST_JOB_3 = "testJob3";
   public static final String TEST_TABLE_NAME_3 = "ijkl";
   private String metaConnect;
   private String metaUser;
   private String metaPassword;
   private String driverClass;
   private JobStorage storage;

   private Configuration conf;
   private Map<String, String> descriptor;

   public SavedJobsTestBase(String metaConnect, String metaUser, String metaPassword, String driverClass){
     this.metaConnect = metaConnect;
     this.metaUser = metaUser;
     this.metaPassword = metaPassword;
     this.driverClass = driverClass;
     this.descriptor = new TreeMap<>();
   }

   @Rule
   public ExpectedException thrown = ExpectedException.none();

   @Before
   public void setUp() throws Exception {
     // Delete db state between tests.
     resetJobSchema();
     conf = newConf();

     descriptor.put(META_CONNECT_KEY, metaConnect);
     descriptor.put(META_USERNAME_KEY, metaUser);
     descriptor.put(META_PASSWORD_KEY, metaPassword);

     JobStorageFactory ssf = new JobStorageFactory(conf);
     storage = ssf.getJobStorage(descriptor);
     storage.open(descriptor);
   }

   @After
   public void tearDown() throws Exception {
     descriptor.clear();
     storage.close();
   }

   public void resetJobSchema()
           throws SQLException {
     SqoopOptions options = new SqoopOptions();
     options.setConnectString(metaConnect);
     options.setUsername(metaUser);
     options.setPassword(metaPassword);
     options.setDriverClassName(driverClass);

     resetSchema(options);
   }

   /**
    * Drop all tables in the configured HSQLDB-based schema/user/pass.
    */
   public static void resetSchema(SqoopOptions options) throws SQLException {
     JobData jd = new JobData();
     jd.setSqoopOptions(options);
     DefaultManagerFactory dmf = new DefaultManagerFactory();
     ConnManager manager = dmf.accept(jd);
     Connection c = manager.getConnection();
     Statement s = c.createStatement();
     try {
       String [] tables = manager.listTables();
       for (String table : tables) {
         if(table.equals("SQOOP_ROOT") || table.equals("SQOOP_SESSIONS")){
           s.execute("DROP TABLE " + manager.escapeTableName(table));
         }
       }

       c.commit();
     } finally {
       s.close();
     }
   }

   public Configuration newConf() {
     Configuration conf = new Configuration();
     conf.set(META_CONNECT_KEY, metaConnect);
     conf.set(META_USERNAME_KEY, metaUser);
     conf.set(META_PASSWORD_KEY, metaPassword);

     return conf;
   }

   @Test
   public void testReadJobDoesExistPasses() throws Exception{
     storage.create(TEST_JOB, createTestJobData(TEST_TABLE_NAME));

     assertEquals("Read did not return job data correctly",
             storage.read(TEST_JOB).getSqoopOptions().getTableName(),
             TEST_TABLE_NAME);
   }

   @Test
   public void testUpdateJob() throws  Exception {
     storage.create(TEST_JOB, createTestJobData(TEST_TABLE_NAME));

     storage.update(TEST_JOB, createTestJobData(TEST_TABLE_NAME_2) );

     assertEquals("Update did not change data correctly",
             storage.read(TEST_JOB).getSqoopOptions().getTableName(),
             TEST_TABLE_NAME_2);
   }

   @Test
   public void testList() throws IOException {
     storage.create(TEST_JOB, createTestJobData(TEST_TABLE_NAME));
     storage.create(TEST_JOB_2, createTestJobData(TEST_TABLE_NAME_2));
     storage.create(TEST_JOB_3, createTestJobData(TEST_TABLE_NAME_3));

     assertThat("List did not return correct job data",
             storage.list(), hasItems(TEST_JOB, TEST_JOB_2, TEST_JOB_3));
   }

   @Test
   public void testCreateSameJob() throws IOException {

     // Job list should start out empty.
     List<String> jobs = storage.list();
     assertEquals("Job list should start out empty", 0, jobs.size());

     // Create a job that displays the version.
     JobData data = new JobData(new SqoopOptions(), new VersionTool());
     storage.create(TEST_JOB, data);

     jobs = storage.list();
     assertEquals("Test Job not created correctly",1, jobs.size());
     assertEquals("Test Job data not returned correctly", TEST_JOB, jobs.get(0));

     try {
       // Try to create that same job name again. This should fail.
       thrown.expect(IOException.class);
       thrown.reportMissingExceptionWithMessage("Expected IOException since job already exists");
       storage.create(TEST_JOB, data);
     } finally {
       jobs = storage.list();
       assertEquals("Incorrect number of jobs present",1, jobs.size());

       // Restore our job, check that it exists.
       JobData outData = storage.read(TEST_JOB);
       assertEquals("Test job does not exist", new VersionTool().getToolName(),
           outData.getSqoopTool().getToolName());
     }
   }

   @Test
   public void testDeleteJob() throws IOException {
     // Job list should start out empty.
     List<String> jobs = storage.list();
     assertEquals("Job List should start out empty", 0, jobs.size());

     // Create a job that displays the version.
     JobData data = new JobData(new SqoopOptions(), new VersionTool());
     storage.create(TEST_JOB, data);

     jobs = storage.list();
     assertEquals("Incorrect number of jobs present",1, jobs.size());
     assertEquals("Test Job created incorrectly", TEST_JOB, jobs.get(0));

     // Now delete the job.
     storage.delete(TEST_JOB);

     // After delete, we should have no jobs.
     jobs = storage.list();
     assertEquals("Job was not deleted correctly", 0, jobs.size());
   }

   @Test
   public void testRestoreNonExistingJob() throws IOException {
       // Try to restore a job that doesn't exist. Watch it fail.
       thrown.expect(IOException.class);
       thrown.reportMissingExceptionWithMessage("Expected IOException since job doesn't exist");
       storage.read("DoesNotExist");
   }

   @Test
   public void testCreateJobWithExtraArgs() throws IOException {

         // Job list should start out empty.
         List<String> jobs = storage.list();
         assertEquals("Job list should start out empty", 0, jobs.size());

         // Create a job with extra args
         SqoopOptions opts = new SqoopOptions();
         String[] args = {"-schema", "test"};
         opts.setExtraArgs(args);
         JobData data = new JobData(opts, new VersionTool());
         storage.create(TEST_JOB, data);

         jobs = storage.list();
         assertEquals("Incorrect number of jobs", 1, jobs.size());
         assertEquals("Job not created properly", TEST_JOB, jobs.get(0));

         // Restore our job, check that it exists.
         JobData outData = storage.read(TEST_JOB);
         assertEquals("Incorrect Tool in Test Job",
                 new VersionTool().getToolName(),
                 outData.getSqoopTool().getToolName());

         String[] storedArgs = outData.getSqoopOptions().getExtraArgs();
         for(int index = 0; index < args.length; ++index) {
             assertEquals(args[index], storedArgs[index]);
         }

         // Now delete the job.
         storage.delete(TEST_JOB);
     }

   @Test
   public void testMultiConnections() throws IOException {

     // Job list should start out empty.
     List<String> jobs = storage.list();
     assertEquals("Job list should start out empty", 0, jobs.size());

     // Create a job that displays the version.
     JobData data = new JobData(new SqoopOptions(), new VersionTool());
     storage.create(TEST_JOB, data);

     jobs = storage.list();
     assertEquals("Incorrect number of jobs", 1, jobs.size());
     assertEquals("Job not created correctly", TEST_JOB, jobs.get(0));

     storage.close(); // Close the existing connection

     // Now re-open the storage.
     storage.open(descriptor);

     jobs = storage.list();
     assertEquals("Test Job did not persist through re-open", 1, jobs.size());
     assertEquals("Job data not correct after re-open", TEST_JOB, jobs.get(0));

     // Restore our job, check that it exists.
     JobData outData = storage.read(TEST_JOB);
     assertEquals("Incorrect Tool in Test Job",
             new VersionTool().getToolName(),
             outData.getSqoopTool().getToolName());
   }

   private org.apache.sqoop.metastore.JobData createTestJobData(String setTableName) throws IOException {
     SqoopOptions testOpts = new SqoopOptions();
     testOpts.setTableName(setTableName);
     ImportTool testTool = new ImportTool();
     return new org.apache.sqoop.metastore.JobData(testOpts,testTool);

   }
 }
	/**
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance
	* with the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/

	package org.apache.sqoop.metastore;

	import static org.apache.sqoop.metastore.GenericJobStorage.META_CONNECT_KEY;
	import static org.apache.sqoop.metastore.GenericJobStorage.META_PASSWORD_KEY;
	import static org.apache.sqoop.metastore.GenericJobStorage.META_USERNAME_KEY;

	import static org.hamcrest.core.IsCollectionContaining.hasItems;
	import static org.junit.Assert.assertEquals;
	import static org.junit.Assert.assertThat;

	import org.apache.sqoop.manager.ConnManager;
	import org.apache.sqoop.SqoopOptions;
	import org.apache.sqoop.tool.VersionTool;

	import org.apache.hadoop.conf.Configuration;
	import org.apache.sqoop.manager.DefaultManagerFactory;
	import org.apache.sqoop.tool.ImportTool;
	import org.apache.sqoop.metastore.JobData;
	import org.apache.sqoop.metastore.JobStorage;
	import org.apache.sqoop.metastore.JobStorageFactory;
	import org.junit.After;
	import org.junit.Before;
	import org.junit.Rule;
	import org.junit.Test;
	import org.junit.rules.ExpectedException;

	import java.io.IOException;

	import java.sql.Connection;
	import java.sql.SQLException;
	import java.sql.Statement;

	import java.util.List;
	import java.util.Map;
	import java.util.TreeMap;

	/**
	* Test the metastore and job-handling features,
	* implemented for specific database services in sub-classes.
	*/
	public abstract class SavedJobsTestBase {

	public static final String TEST_JOB = "testJob";
	public static final String TEST_TABLE_NAME = "abcd";
	public static final String TEST_TABLE_NAME_2 = "efgh";
	public static final String TEST_JOB_2 = "testJob2";
	public static final String TEST_JOB_3 = "testJob3";
	public static final String TEST_TABLE_NAME_3 = "ijkl";
	private String metaConnect;
	private String metaUser;
	private String metaPassword;
	private String driverClass;
	private JobStorage storage;

	private Configuration conf;
	private Map<String, String> descriptor;

	public SavedJobsTestBase(String metaConnect, String metaUser, String metaPassword, String driverClass){
	this.metaConnect = metaConnect;
	this.metaUser = metaUser;
	this.metaPassword = metaPassword;
	this.driverClass = driverClass;
	this.descriptor = new TreeMap<>();
	}

	@Rule
	public ExpectedException thrown = ExpectedException.none();

	@Before
	public void setUp() throws Exception {
	// Delete db state between tests.
	resetJobSchema();
	conf = newConf();

	descriptor.put(META_CONNECT_KEY, metaConnect);
	descriptor.put(META_USERNAME_KEY, metaUser);
	descriptor.put(META_PASSWORD_KEY, metaPassword);

	JobStorageFactory ssf = new JobStorageFactory(conf);
	storage = ssf.getJobStorage(descriptor);
	storage.open(descriptor);
	}

	@After
	public void tearDown() throws Exception {
	descriptor.clear();
	storage.close();
	}

	public void resetJobSchema()
	throws SQLException {
	SqoopOptions options = new SqoopOptions();
	options.setConnectString(metaConnect);
	options.setUsername(metaUser);
	options.setPassword(metaPassword);
	options.setDriverClassName(driverClass);

	resetSchema(options);
	}

	/**
	* Drop all tables in the configured HSQLDB-based schema/user/pass.
	*/
	public static void resetSchema(SqoopOptions options) throws SQLException {
	JobData jd = new JobData();
	jd.setSqoopOptions(options);
	DefaultManagerFactory dmf = new DefaultManagerFactory();
	ConnManager manager = dmf.accept(jd);
	Connection c = manager.getConnection();
	Statement s = c.createStatement();
	try {
	String [] tables = manager.listTables();
	for (String table : tables) {
	if(table.equals("SQOOP_ROOT") \|\| table.equals("SQOOP_SESSIONS")){
	s.execute("DROP TABLE " + manager.escapeTableName(table));
	}
	}

	c.commit();
	} finally {
	s.close();
	}
	}

	public Configuration newConf() {
	Configuration conf = new Configuration();
	conf.set(META_CONNECT_KEY, metaConnect);
	conf.set(META_USERNAME_KEY, metaUser);
	conf.set(META_PASSWORD_KEY, metaPassword);

	return conf;
	}

	@Test
	public void testReadJobDoesExistPasses() throws Exception{
	storage.create(TEST_JOB, createTestJobData(TEST_TABLE_NAME));

	assertEquals("Read did not return job data correctly",
	storage.read(TEST_JOB).getSqoopOptions().getTableName(),
	TEST_TABLE_NAME);
	}

	@Test
	public void testUpdateJob() throws Exception {
	storage.create(TEST_JOB, createTestJobData(TEST_TABLE_NAME));

	storage.update(TEST_JOB, createTestJobData(TEST_TABLE_NAME_2) );

	assertEquals("Update did not change data correctly",
	storage.read(TEST_JOB).getSqoopOptions().getTableName(),
	TEST_TABLE_NAME_2);
	}

	@Test
	public void testList() throws IOException {
	storage.create(TEST_JOB, createTestJobData(TEST_TABLE_NAME));
	storage.create(TEST_JOB_2, createTestJobData(TEST_TABLE_NAME_2));
	storage.create(TEST_JOB_3, createTestJobData(TEST_TABLE_NAME_3));

	assertThat("List did not return correct job data",
	storage.list(), hasItems(TEST_JOB, TEST_JOB_2, TEST_JOB_3));
	}

	@Test
	public void testCreateSameJob() throws IOException {

	// Job list should start out empty.
	List<String> jobs = storage.list();
	assertEquals("Job list should start out empty", 0, jobs.size());

	// Create a job that displays the version.
	JobData data = new JobData(new SqoopOptions(), new VersionTool());
	storage.create(TEST_JOB, data);

	jobs = storage.list();
	assertEquals("Test Job not created correctly",1, jobs.size());
	assertEquals("Test Job data not returned correctly", TEST_JOB, jobs.get(0));

	try {
	// Try to create that same job name again. This should fail.
	thrown.expect(IOException.class);
	thrown.reportMissingExceptionWithMessage("Expected IOException since job already exists");
	storage.create(TEST_JOB, data);
	} finally {
	jobs = storage.list();
	assertEquals("Incorrect number of jobs present",1, jobs.size());

	// Restore our job, check that it exists.
	JobData outData = storage.read(TEST_JOB);
	assertEquals("Test job does not exist", new VersionTool().getToolName(),
	outData.getSqoopTool().getToolName());
	}
	}

	@Test
	public void testDeleteJob() throws IOException {
	// Job list should start out empty.
	List<String> jobs = storage.list();
	assertEquals("Job List should start out empty", 0, jobs.size());

	// Create a job that displays the version.
	JobData data = new JobData(new SqoopOptions(), new VersionTool());
	storage.create(TEST_JOB, data);

	jobs = storage.list();
	assertEquals("Incorrect number of jobs present",1, jobs.size());
	assertEquals("Test Job created incorrectly", TEST_JOB, jobs.get(0));

	// Now delete the job.
	storage.delete(TEST_JOB);

	// After delete, we should have no jobs.
	jobs = storage.list();
	assertEquals("Job was not deleted correctly", 0, jobs.size());
	}

	@Test
	public void testRestoreNonExistingJob() throws IOException {
	// Try to restore a job that doesn't exist. Watch it fail.
	thrown.expect(IOException.class);
	thrown.reportMissingExceptionWithMessage("Expected IOException since job doesn't exist");
	storage.read("DoesNotExist");
	}

	@Test
	public void testCreateJobWithExtraArgs() throws IOException {

	// Job list should start out empty.
	List<String> jobs = storage.list();
	assertEquals("Job list should start out empty", 0, jobs.size());

	// Create a job with extra args
	SqoopOptions opts = new SqoopOptions();
	String[] args = {"-schema", "test"};
	opts.setExtraArgs(args);
	JobData data = new JobData(opts, new VersionTool());
	storage.create(TEST_JOB, data);

	jobs = storage.list();
	assertEquals("Incorrect number of jobs", 1, jobs.size());
	assertEquals("Job not created properly", TEST_JOB, jobs.get(0));

	// Restore our job, check that it exists.
	JobData outData = storage.read(TEST_JOB);
	assertEquals("Incorrect Tool in Test Job",
	new VersionTool().getToolName(),
	outData.getSqoopTool().getToolName());

	String[] storedArgs = outData.getSqoopOptions().getExtraArgs();
	for(int index = 0; index < args.length; ++index) {
	assertEquals(args[index], storedArgs[index]);
	}

	// Now delete the job.
	storage.delete(TEST_JOB);
	}

	@Test
	public void testMultiConnections() throws IOException {

	// Job list should start out empty.
	List<String> jobs = storage.list();
	assertEquals("Job list should start out empty", 0, jobs.size());

	// Create a job that displays the version.
	JobData data = new JobData(new SqoopOptions(), new VersionTool());
	storage.create(TEST_JOB, data);

	jobs = storage.list();
	assertEquals("Incorrect number of jobs", 1, jobs.size());
	assertEquals("Job not created correctly", TEST_JOB, jobs.get(0));

	storage.close(); // Close the existing connection

	// Now re-open the storage.
	storage.open(descriptor);

	jobs = storage.list();
	assertEquals("Test Job did not persist through re-open", 1, jobs.size());
	assertEquals("Job data not correct after re-open", TEST_JOB, jobs.get(0));

	// Restore our job, check that it exists.
	JobData outData = storage.read(TEST_JOB);
	assertEquals("Incorrect Tool in Test Job",
	new VersionTool().getToolName(),
	outData.getSqoopTool().getToolName());
	}

	private org.apache.sqoop.metastore.JobData createTestJobData(String setTableName) throws IOException {
	SqoopOptions testOpts = new SqoopOptions();
	testOpts.setTableName(setTableName);
	ImportTool testTool = new ImportTool();
	return new org.apache.sqoop.metastore.JobData(testOpts,testTool);

	}
	}