public CompatibilityInfoMap checkPegasusSchemaCompatibility(String prevPegasusSchemaPath, String currentPegasusSchemaPath,
      CompatibilityOptions.Mode compatMode)
  {
    boolean newSchemaCreated = false;
    boolean preSchemaRemoved = false;

    DataSchema preSchema = null;
    try
    {
      preSchema = parseSchema(new File(prevPegasusSchemaPath));
    }
    catch(FileNotFoundException e)
    {
      newSchemaCreated = true;
    }

    DataSchema currSchema = null;
    try
    {
      currSchema = parseSchema(new File(currentPegasusSchemaPath));
    }
    catch(FileNotFoundException e)
    {
      preSchemaRemoved = true;
    }

    if (newSchemaCreated && !preSchemaRemoved)
    {
      constructCompatibilityMessage(CompatibilityMessage.Impact.NEW_SCHEMA_ADDED,
          "New schema %s is created.", currentPegasusSchemaPath);
    }
    if (!newSchemaCreated && preSchemaRemoved)
    {
      constructCompatibilityMessage(CompatibilityMessage.Impact.BREAK_OLD_CLIENTS,
          "Schema %s is removed.", prevPegasusSchemaPath);
    }

    if (preSchema == null || currSchema == null)
    {
      return _infoMap;
    }

    CompatibilityOptions compatibilityOptions = new CompatibilityOptions().setMode(compatMode).setAllowPromotions(true);
    CompatibilityResult result = CompatibilityChecker.checkCompatibility(preSchema, currSchema, compatibilityOptions);

    if (!result.getMessages().isEmpty())
    {
      result.getMessages().forEach(message -> _infoMap.addModelInfo(message));
    }

    if (!_handlers.isEmpty())
    {
      List<SchemaAnnotationHandler.AnnotationCompatibilityResult> annotationCompatibilityResults =
          AnnotationCompatibilityChecker.checkPegasusSchemaAnnotation(preSchema, currSchema, _handlers);
      for (SchemaAnnotationHandler.AnnotationCompatibilityResult annotationResult: annotationCompatibilityResults)
      {
        if (!annotationResult.getMessages().isEmpty())
        {
          annotationResult.getMessages().forEach(message -> _infoMap.addAnnotation(message));
        }
      }
    }

    return _infoMap;
  }
@Test(dataProvider = "incompatibleInputFiles")
  public void testIncompatiblePegasusSchemaSnapshot(String prevSchema, String currSchema,
      Collection<CompatibilityInfo> expectedIncompatibilityErrors, Collection<CompatibilityInfo> expectedWireCompatibilityDiffs,
      Collection<CompatibilityInfo> expectedCompatibilityDiffs )
  {
    PegasusSchemaSnapshotCompatibilityChecker checker = new PegasusSchemaSnapshotCompatibilityChecker();
    CompatibilityInfoMap infoMap = checker.checkPegasusSchemaCompatibility(snapshotDir + FS + prevSchema, snapshotDir + FS + currSchema,
        CompatibilityOptions.Mode.DATA);
    Assert.assertFalse(infoMap.isModelCompatible(CompatibilityLevel.BACKWARDS));
    Assert.assertFalse(infoMap.isModelCompatible(CompatibilityLevel.EQUIVALENT));
    Assert.assertTrue(infoMap.isModelCompatible(CompatibilityLevel.IGNORE));

    final Collection<CompatibilityInfo> modelIncompatibles = infoMap.getModelIncompatibles();
    final Collection<CompatibilityInfo> modelWireCompatibles = infoMap.getModelWireCompatibles();
    final Collection<CompatibilityInfo> modelCompatibles = infoMap.getModelCompatibles();

    for (CompatibilityInfo error : expectedIncompatibilityErrors)
    {
      Assert.assertTrue(modelIncompatibles.contains(error), "Reported model incompatibles should contain: " + error.toString());
      modelIncompatibles.remove(error);
    }
    for (CompatibilityInfo diff : expectedWireCompatibilityDiffs)
    {
      Assert.assertTrue(modelWireCompatibles.contains(diff), "Reported model wireCompatibles should contain: " + diff.toString());
      modelWireCompatibles.remove(diff);
    }
    for (CompatibilityInfo diff : expectedCompatibilityDiffs)
    {
      Assert.assertTrue(modelCompatibles.contains(diff), "Reported model compatibles should contain: " + diff.toString());
      modelCompatibles.remove(diff);
    }

    Assert.assertTrue(modelIncompatibles.isEmpty());
    Assert.assertTrue(modelWireCompatibles.isEmpty());
    Assert.assertTrue(modelCompatibles.isEmpty());
  }