testresultsettoxmlconverter.java

java开源的企业总线.xmlBlaster

/*------------------------------------------------------------------------------
Name:      TestResultSetToXmlConverter.java
Project:   org.xmlBlasterProject:   xmlBlaster.org
Copyright: xmlBlaster.org, see xmlBlaster-LICENSE file
------------------------------------------------------------------------------*/
package org.xmlBlaster.test.contrib.dbwatcher;

import java.io.IOException;
import java.io.InputStream;
import java.util.prefs.Preferences;


import junit.framework.TestCase;
import org.custommonkey.xmlunit.XMLTestCase;
import org.xmlBlaster.contrib.I_Info;
import org.xmlBlaster.contrib.I_Update;
import org.xmlBlaster.contrib.db.DbPool;
import org.xmlBlaster.contrib.db.I_DbPool;
import org.xmlBlaster.contrib.dbwatcher.DbWatcher;
import org.xmlBlaster.contrib.dbwatcher.Info;
import org.xmlBlaster.contrib.dbwatcher.detector.I_ChangeDetector;
import org.xmlBlaster.contrib.dbwatcher.mom.XmlBlasterPublisher;
import org.xmlBlaster.test.contrib.TestUtils;

import java.util.logging.Logger;

import java.util.Map;
import java.util.HashMap;

/**
 * Test basic functionality. 
 * <p> 
 * To run most of the tests you need to have a databse (for example Oracle)
 * and XmlBlaster up and running.
 * </p>
 * <p>
 * The connection configuration (url, password etc.) is configured
 * as JVM property or in {@link #createTest(I_Info, Map)} and
 * {@link #setUpDbPool(I_Info)}
 * </p> 
 *
 * @see DbWatcher
 * @author Marcel Ruff
 */
public class TestResultSetToXmlConverter extends XMLTestCase {
    private static Logger log = Logger.getLogger(TestResultSetToXmlConverter.class.getName());
    private Preferences prefs;
    private I_Info info;
    private I_DbPool dbPool;
    private Map updateMap = new HashMap(); // collects received update messages
    private DbWatcher processor;

    /**
     * Start the test. 
     * <pre>
     * java -Ddb.password=secret junit.swingui.TestRunner -noloading org.xmlBlaster.test.contrib.dbwatcher.TestResultSetToXmlConverter
     * </pre>
     * @param args Command line settings
     */
    public static void main(String[] args) {
        junit.swingui.TestRunner.run(TestResultSetToXmlConverter.class);
    }

    /**
     * Default ctor. 
     */
    public TestResultSetToXmlConverter() {
    }

   /**
    * Constructor for TestResultSetToXmlConverter.
    * @param arg0
    */
    public TestResultSetToXmlConverter(String arg0) {
       super(arg0);
    }

    /**
     * Configure database access. 
     * @see TestCase#setUp()
     */
   protected void setUp() throws Exception {
      super.setUp();
      this.prefs = Preferences.userRoot();
      this.prefs.clear();
      this.info = new Info(this.prefs);
      
      this.dbPool = setUpDbPool(info);
      try {
         this.dbPool.update("DROP TABLE TEST_POLL");
      } catch(Exception e) {
         log.warning(e.toString()); 
      }
      
      this.processor = null;
   }
   
   /**
    * Creates a database pooling instance and puts it to info. 
    * @param info The configuration
    * @return The created pool
    */
   public static DbPool setUpDbPool(I_Info info) {
      String driverClass = System.getProperty("jdbc.drivers", "org.hsqldb.jdbcDriver:oracle.jdbc.driver.OracleDriver:com.microsoft.jdbc.sqlserver.SQLServerDriver:org.postgresql.Driver");
      ////System.setProperty("jdbc.drivers", driverClass);

      /*
      String dbUrl = System.getProperty("db.url", "jdbc:oracle:thin:@localhost:1521:orcl");
      String dbUser = System.getProperty("db.user", "system");
      String dbPassword = System.getProperty("db.password", "");
      */
      
      String dbUrl = System.getProperty("db.url", "jdbc:oracle:thin:@desktop:1521:test");
      String dbUser = System.getProperty("db.user", "system");
      String dbPassword = System.getProperty("db.password", "frifra20");
      
      //String fs = System.getProperty("file.separator");
      //String dbUrl = "jdbc:hsqldb:"+System.getProperty("user.home")+fs+"tmp"+fs+"testpoll";
      //String dbUser = "sa";
      //String dbPassword = "";

      info.put("jdbc.drivers", driverClass);
      info.put("db.url", dbUrl);
      info.put("db.user", dbUser);
      info.put("db.password", dbPassword);
        
      DbPool dbPool = new DbPool();
      dbPool.init(info);
      info.putObject("db.pool", dbPool);
      
      return dbPool;
   }

   /**
    * Creates a DbWatcher instance and listens on MoM messages. 
    * @param info Configuration
    * @param updateMap The map for received messages
    * @return A new DbWatcher
    * @throws Exception
    */
   public static DbWatcher createTest(I_Info info, final Map updateMap) throws Exception {
      /*
      // Configure the MoM
      this.prefs.put("mom.connectQos", 
                     "<qos>" +
                     " <securityService type='htpasswd' version='1.0'>" +
                     "   <![CDATA[" + 
                     "   <user>michele</user>" +
                     "   <passwd>secret</passwd>" +
                     "   ]]>" +
                     " </securityService>" +
                     " <session name='joe/3'/>'" +
                     " <address type='SOCKET'>" +
                     "   socket://192.168.110.10:7607" +
                     " </address>" +
                     " </qos>");
      System.setProperty("protocol", "SOCKET");
      System.setProperty("protocol/socket/hostname", "192.168.110.10");
      */

      DbWatcher pc = new DbWatcher(info);
      XmlBlasterPublisher mom = (XmlBlasterPublisher)pc.getMom();
      mom.subscribe("XPATH://key", new I_Update() {
         public void update(String topic, java.io.InputStream is, Map attrMap) {
            log.info("Received '" + topic + "' from MoM");
            try {
               writeToFile(topic, new String(TestUtils.getContent(is)));
            }
            catch (Exception e) {
               // Ignore   
            }
            updateMap.put(topic, is);
         }
      });
      
      try { Thread.sleep(1000); } catch(Exception e) { /* Ignore */ }
      updateMap.clear(); // Ignore any existing topics

      pc.startAlertProducers();
      
      return pc;
   }

   /*
    * @see TestCase#tearDown()
    */
   protected void tearDown() throws Exception {
      super.tearDown();
       
      if (this.processor != null) {
         this.processor.shutdown();
         this.processor = null;
      }
       
      if (this.dbPool != null) {
         try {
            this.dbPool.update("DROP TABLE TEST_POLL");
         } catch(Exception e) {
            log.warning(e.toString()); 
         }
         this.dbPool.shutdown();
      }
   }

   /**
    * If the table does not exist we expect a null ResultSet
    * @throws Exception Any type is possible
    */
   public final void testTableStates() throws Exception {
      log.info("Start testTableStates()");

      this.prefs.put("converter.rootName", "myRootTag");
      this.prefs.put("db.queryMeatStatement", "");
      this.prefs.put("alertScheduler.pollInterval", "0"); // switch off
      this.prefs.put("changeDetector.groupColName", ""); // !!! Tests without grouping
      this.prefs.put("converter.addMeta", ""+true);
      this.prefs.put("changeDetector.detectStatement", "SELECT colKey, col1, col2, ICAO_ID FROM TEST_POLL");
      this.prefs.put("mom.topicName", "db.change.event.TEST_POLL");
      
      this.processor = createTest(new Info(prefs), this.updateMap);
      I_ChangeDetector changeDetector = processor.getChangeDetector();
      
      for (int i=0; i<2; i++) {
         log.info("Testing no table ...");
         changeDetector.checkAgain(null);
         sleep(500);
         assertEquals("Number of message is wrong", 0, this.updateMap.size());
      }

      {
      log.info("Now testing an empty table ...");
      this.dbPool.update("CREATE TABLE TEST_POLL (colKey NUMBER(10,3), col1 VARCHAR(20), col2 NUMBER(12), ICAO_ID VARCHAR(10))");
      changeDetector.checkAgain(null);
      sleep(500);
      assertEquals("Number of message is wrong", 1, this.updateMap.size());
      String xml = (String)this.updateMap.get("db.change.event.TEST_POLL");
      assertNotNull("No db.change.event.${groupColValue} message has arrived", xml);
      assertXpathNotExists("/myRootTag/row[@num='0']", xml);
      assertXpathEvaluatesTo("CREATE", "/myRootTag/desc/command/text()", xml);
      this.updateMap.clear();

      writeToFile("db.change.event.CREATE", xml);

      changeDetector.checkAgain(null);
      sleep(500);
      assertEquals("Number of message is wrong", 0, this.updateMap.size());
      }

      {
      log.info("Insert one row");
      this.dbPool.update("INSERT INTO TEST_POLL VALUES ('1.1', '<Bla', '9000', 'EDDI')");
      changeDetector.checkAgain(null);
      sleep(500);
      assertEquals("Number of message is wrong", 1, this.updateMap.size());
      String xml = (String)this.updateMap.get("db.change.event.TEST_POLL");
      assertNotNull("xml returned is null", xml);
      // TODO: We deliver a "UPDATE" because of the CREATE md5: Is it easy possible to detect the INSERT?
      assertXpathEvaluatesTo("UPDATE", "/myRootTag/desc/command/text()", xml);
      assertXpathEvaluatesTo("<Bla", "/myRootTag/row[@num='0']/col[@name='COL1']/text()", xml);
      this.updateMap.clear();

      writeToFile("db.change.event.INSERT", xml);

      changeDetector.checkAgain(null);
      sleep(500);
      assertEquals("Number of message is wrong", 0, this.updateMap.size());
      }
            
      {
         log.info("Update one row");
         this.dbPool.update("UPDATE TEST_POLL SET col1='BXXX' WHERE ICAO_ID='EDDI'");
         changeDetector.checkAgain(null);
         sleep(500);
         assertEquals("Number of message is wrong", 1, this.updateMap.size());
         String xml = (String)this.updateMap.get("db.change.event.TEST_POLL");
         assertXpathEvaluatesTo("UPDATE", "/myRootTag/desc/command/text()", xml);
         assertXpathEvaluatesTo("BXXX", "/myRootTag/row[@num='0']/col[@name='COL1']/text()", xml);
         this.updateMap.clear();

         writeToFile("db.change.event.UPDATE", xml);

         changeDetector.checkAgain(null);
         sleep(500);
         assertEquals("Number of message is wrong", 0, this.updateMap.size());
      }

      {
         log.info("Delete one row");
         this.dbPool.update("DELETE FROM TEST_POLL WHERE ICAO_ID='EDDI'");
         changeDetector.checkAgain(null);
         sleep(500);
         assertEquals("Number of message is wrong", 1, this.updateMap.size());
         String xml = (String)this.updateMap.get("db.change.event.TEST_POLL");
         // TODO: We deliver "UPDATE" instead of DELETE:
         assertXpathEvaluatesTo("UPDATE", "/myRootTag/desc/command/text()", xml);
         assertXpathNotExists("/myRootTag/row[@num='0']", xml);
         this.updateMap.clear();

         writeToFile("db.change.event.DELETE", xml);

         changeDetector.checkAgain(null);
         sleep(500);
         assertEquals("Number of message is wrong", 0, this.updateMap.size());
      }

      {
         log.info("Drop a table");