@Test
public void testExecuteDDLUpdates() throws Exception {

  final DBAccessor dbAccessor = createNiceMock(DBAccessor.class);
  Configuration configuration = createNiceMock(Configuration.class);
  Connection connection = createNiceMock(Connection.class);
  Statement statement = createNiceMock(Statement.class);
  ResultSet resultSet = createNiceMock(ResultSet.class);
  expect(configuration.getDatabaseUrl()).andReturn(Configuration.JDBC_IN_MEMORY_URL).anyTimes();
  Capture<DBAccessor.DBColumnInfo> alertDefinitionIgnoreColumnCapture = EasyMock.newCapture();
  Capture<DBAccessor.DBColumnInfo> alertDefinitionDescriptionColumnCapture = EasyMock.newCapture();
  Capture<DBAccessor.DBColumnInfo> alertTargetGlobalColumnCapture = EasyMock.newCapture();
  Capture<DBAccessor.DBColumnInfo> hostComponentStateColumnCapture = EasyMock.newCapture();
  Capture<DBAccessor.DBColumnInfo> hostComponentVersionColumnCapture = EasyMock.newCapture();
  Capture<DBAccessor.DBColumnInfo> clustersSecurityTypeColumnCapture = EasyMock.newCapture();
  Capture<DBAccessor.DBColumnInfo> hostComponentStateSecurityStateColumnCapture = EasyMock.newCapture();
  Capture<DBAccessor.DBColumnInfo> hostComponentDesiredStateSecurityStateColumnCapture = EasyMock.newCapture();
  Capture<DBAccessor.DBColumnInfo> hostRoleCommandRetryColumnCapture = EasyMock.newCapture();
  Capture<DBAccessor.DBColumnInfo> stageSkippableColumnCapture = EasyMock.newCapture();
  Capture<DBAccessor.DBColumnInfo> viewparameterLabelColumnCapture = EasyMock.newCapture();
  Capture<DBAccessor.DBColumnInfo> viewparameterPlaceholderColumnCapture = EasyMock.newCapture();
  Capture<DBAccessor.DBColumnInfo> viewparameterDefaultValueColumnCapture = EasyMock.newCapture();
  Capture<DBAccessor.DBColumnInfo> serviceDesiredStateSecurityStateColumnCapture = EasyMock.newCapture();
  Capture<List<DBAccessor.DBColumnInfo>> clusterVersionCapture = EasyMock.newCapture();
  Capture<List<DBAccessor.DBColumnInfo>> hostVersionCapture = EasyMock.newCapture();
  Capture<DBAccessor.DBColumnInfo> valueColumnCapture = EasyMock.newCapture();
  Capture<DBAccessor.DBColumnInfo> dataValueColumnCapture = EasyMock.newCapture();
  Capture<List<DBAccessor.DBColumnInfo>> alertTargetStatesCapture = EasyMock.newCapture();
  Capture<List<DBAccessor.DBColumnInfo>> artifactCapture = EasyMock.newCapture();
  Capture<List<DBAccessor.DBColumnInfo>> kerberosPrincipalCapture = EasyMock.newCapture();
  Capture<List<DBAccessor.DBColumnInfo>> kerberosPrincipalHostCapture = EasyMock.newCapture();
  Capture<List<DBAccessor.DBColumnInfo>> upgradeCapture = EasyMock.newCapture();
  Capture<List<DBAccessor.DBColumnInfo>> upgradeGroupCapture = EasyMock.newCapture();
  Capture<List<DBAccessor.DBColumnInfo>> upgradeItemCapture = EasyMock.newCapture();
  // Alert Definition
  dbAccessor.addColumn(eq("alert_definition"),
                        capture(alertDefinitionIgnoreColumnCapture));
  dbAccessor.addColumn(eq("alert_definition"),
      capture(alertDefinitionDescriptionColumnCapture));
  dbAccessor.createTable(eq("alert_target_states"),
      capture(alertTargetStatesCapture));
  // alert target
  dbAccessor.addColumn(eq("alert_target"),
      capture(alertTargetGlobalColumnCapture));
  // Host Component State
  dbAccessor.addColumn(eq("hostcomponentstate"),
      capture(hostComponentStateColumnCapture));
  // Host Component Version
  dbAccessor.addColumn(eq("hostcomponentstate"),
      capture(hostComponentVersionColumnCapture));
  // Host Role Command retry allowed
  dbAccessor.addColumn(eq("host_role_command"),
      capture(hostRoleCommandRetryColumnCapture));
  // Stage skippable
  dbAccessor.addColumn(eq("stage"),
      capture(stageSkippableColumnCapture));
  // Clusters: security type
  dbAccessor.addColumn(eq("clusters"),
      capture(clustersSecurityTypeColumnCapture));
  // Host Component State: security State
  dbAccessor.addColumn(eq("hostcomponentstate"),
      capture(hostComponentStateSecurityStateColumnCapture));
  // Host Component Desired State: security State
  dbAccessor.addColumn(eq("hostcomponentdesiredstate"),
      capture(hostComponentDesiredStateSecurityStateColumnCapture));
  dbAccessor.addColumn(eq("viewparameter"), capture(viewparameterLabelColumnCapture));
  dbAccessor.addColumn(eq("viewparameter"), capture(viewparameterPlaceholderColumnCapture));
  dbAccessor.addColumn(eq("viewparameter"), capture(viewparameterDefaultValueColumnCapture));
  // Service Desired State: security State
  dbAccessor.addColumn(eq("servicedesiredstate"),
      capture(serviceDesiredStateSecurityStateColumnCapture));
  // Cluster Version
  dbAccessor.createTable(eq("cluster_version"),
      capture(clusterVersionCapture), eq("id"));
  // Host Version
  dbAccessor.createTable(eq("host_version"),
      capture(hostVersionCapture), eq("id"));
  // Upgrade
  dbAccessor.createTable(eq("upgrade"), capture(upgradeCapture), eq("upgrade_id"));
  // Upgrade Group item
  dbAccessor.createTable(eq("upgrade_group"), capture(upgradeGroupCapture), eq("upgrade_group_id"));
  // Upgrade item
  dbAccessor.createTable(eq("upgrade_item"), capture(upgradeItemCapture), eq("upgrade_item_id"));
  // artifact
  dbAccessor.createTable(eq("artifact"), capture(artifactCapture),
      eq("artifact_name"), eq("foreign_keys"));
  // kerberos_principal
  dbAccessor.createTable(eq("kerberos_principal"), capture(kerberosPrincipalCapture),
      eq("principal_name"));
  // kerberos_principal_host
  dbAccessor.createTable(eq("kerberos_principal_host"), capture(kerberosPrincipalHostCapture),
      eq("principal_name"), eq("host_name"));
  expect(dbAccessor.tableHasColumn("kerberos_principal_host", "host_name")).andReturn(true).atLeastOnce();
  dbAccessor.addFKConstraint(eq("kerberos_principal_host"), eq("FK_krb_pr_host_hostname"),
      eq("host_name"), eq("hosts"), eq("host_name"), eq(true), eq(false));
  dbAccessor.addFKConstraint(eq("kerberos_principal_host"), eq("FK_krb_pr_host_principalname"),
      eq("principal_name"), eq("kerberos_principal"), eq("principal_name"), eq(true), eq(false));
  setViewInstancePropertyExpectations(dbAccessor, valueColumnCapture);
  setViewInstanceDataExpectations(dbAccessor, dataValueColumnCapture);
  // AbstractUpgradeCatalog.addSequence()
  dbAccessor.getConnection();
  expectLastCall().andReturn(connection).anyTimes();
  connection.createStatement();
  expectLastCall().andReturn(statement).anyTimes();
  statement.executeQuery(anyObject(String.class));
  expectLastCall().andReturn(resultSet).anyTimes();
  replay(dbAccessor, configuration, resultSet, statement, connection);
  AbstractUpgradeCatalog upgradeCatalog = getUpgradeCatalog(dbAccessor);
  Class<?> c = AbstractUpgradeCatalog.class;
  Field f = c.getDeclaredField("configuration");
  f.setAccessible(true);
  f.set(upgradeCatalog, configuration);
  upgradeCatalog.executeDDLUpdates();
  verify(dbAccessor, configuration, resultSet, statement, connection);
  // verify columns for alert_definition
  verifyAlertDefinitionIgnoreColumn(alertDefinitionIgnoreColumnCapture);
  verifyAlertDefinitionDescriptionColumn(alertDefinitionDescriptionColumnCapture);
  // verify alert target column for is_global
  verifyAlertTargetGlobal(alertTargetGlobalColumnCapture);
  // verify new table for alert target states
  verifyAlertTargetStatesTable(alertTargetStatesCapture);
  // Verify added column in hostcomponentstate table
  DBAccessor.DBColumnInfo upgradeStateColumn = hostComponentStateColumnCapture.getValue();
  assertEquals("upgrade_state", upgradeStateColumn.getName());
  assertEquals(32, (int) upgradeStateColumn.getLength());
  assertEquals(String.class, upgradeStateColumn.getType());
  assertEquals("NONE", upgradeStateColumn.getDefaultValue());
  assertFalse(upgradeStateColumn.isNullable());
  // Verify added column in hostcomponentstate table
  DBAccessor.DBColumnInfo upgradeVersionColumn = hostComponentVersionColumnCapture.getValue();
  assertEquals("version", upgradeVersionColumn.getName());
  assertEquals(32, (int) upgradeVersionColumn.getLength());
  assertEquals(String.class, upgradeVersionColumn.getType());
  assertEquals("UNKNOWN", upgradeVersionColumn.getDefaultValue());
  assertFalse(upgradeVersionColumn.isNullable());
  // Verify added column in host_role_command table
  DBAccessor.DBColumnInfo upgradeRetryColumn = hostRoleCommandRetryColumnCapture.getValue();
  assertEquals("retry_allowed", upgradeRetryColumn.getName());
  assertEquals(1, (int) upgradeRetryColumn.getLength());
  assertEquals(Integer.class, upgradeRetryColumn.getType());
  assertEquals(0, upgradeRetryColumn.getDefaultValue());
  assertFalse(upgradeRetryColumn.isNullable());
  // Verify added column in host_role_command table
  DBAccessor.DBColumnInfo upgradeSkippableColumn = stageSkippableColumnCapture.getValue();
  assertEquals("skippable", upgradeSkippableColumn.getName());
  assertEquals(1, (int) upgradeSkippableColumn.getLength());
  assertEquals(Integer.class, upgradeSkippableColumn.getType());
  assertEquals(0, upgradeSkippableColumn.getDefaultValue());
  assertFalse(upgradeSkippableColumn.isNullable());
  // verify security_type column
  verifyClustersSecurityType(clustersSecurityTypeColumnCapture);
  // verify security_state columns
  verifyComponentSecurityStateColumn(hostComponentStateSecurityStateColumnCapture);
  verifyComponentSecurityStateColumn(hostComponentDesiredStateSecurityStateColumnCapture);
  verifyServiceSecurityStateColumn(serviceDesiredStateSecurityStateColumnCapture);
  verifyViewParameterColumns(viewparameterLabelColumnCapture, viewparameterPlaceholderColumnCapture,
      viewparameterDefaultValueColumnCapture);
  // verify artifact columns
  List<DBAccessor.DBColumnInfo> artifactColumns = artifactCapture.getValue();
  testCreateArtifactTable(artifactColumns);
  // verify kerberos_principal columns
  testCreateKerberosPrincipalTable(kerberosPrincipalCapture.getValue());
  // verify kerberos_principal_host columns
  testCreateKerberosPrincipalHostTable(kerberosPrincipalHostCapture.getValue());
  // Verify capture group sizes
  assertEquals(7, clusterVersionCapture.getValue().size());
  assertEquals(4, hostVersionCapture.getValue().size());
  assertViewInstancePropertyColumns(valueColumnCapture);
  assertViewInstanceDataColumns(dataValueColumnCapture);
  assertEquals(6, upgradeCapture.getValue().size());
  assertEquals(4, upgradeGroupCapture.getValue().size());
  assertEquals(7, upgradeItemCapture.getValue().size());
}