### Question:
HdfsSnapshotLocationManager implements SourceLocationManager { @Override public Path getPartitionSubPath(Path partitionLocation) { String sourceDataPathString = sourceDataPath.toString(); String partitionLocationString = partitionLocation.toString(); if (!partitionLocationString.startsWith(sourceDataPathString)) { throw new CircusTrainException("Partition path '" + partitionLocationString + "' does not seem to belong to data source path '" + sourceDataPathString + "'"); } String subPath = partitionLocationString.replace(sourceDataPathString, ""); if (subPath.charAt(0) == '/') { subPath = subPath.substring(1); } return new Path(subPath); } HdfsSnapshotLocationManager(
      HiveConf sourceHiveConf,
      String eventId,
      Table sourceTable,
      boolean snapshotsDisabled,
      SourceCatalogListener sourceCatalogListener);   HdfsSnapshotLocationManager(
      HiveConf sourceHiveConf,
      String eventId,
      Table sourceTable,
      boolean snapshotsDisabled,
      String tableBasePath,
      SourceCatalogListener sourceCatalogListener);   HdfsSnapshotLocationManager(
      HiveConf sourceHiveConf,
      String eventId,
      Table sourceTable,
      List<Partition> sourcePartitions,
      boolean snapshotsDisabled,
      String tableBasePath,
      SourceCatalogListener sourceCatalogListener);   HdfsSnapshotLocationManager(
      HiveConf sourceHiveConf,
      String eventId,
      Table sourceTable,
      List<Partition> sourcePartitions,
      boolean snapshotsDisabled,
      String tableBasePath,
      FileSystemFactory fileSystemFactory,
      SourceCatalogListener sourceCatalogListener); @Override Path getTableLocation(); @Override void cleanUpLocations(); @Override List<Path> getPartitionLocations(); @Override Path getPartitionSubPath(Path partitionLocation);  }### Answer:
@Test public void partitionSubPath() throws Exception { StorageDescriptor sd = new StorageDescriptor(); sd.setLocation(PARTITION_BASE_LOCATION + "/partition1"); partition1.setSd(sd); HdfsSnapshotLocationManager manager = new HdfsSnapshotLocationManager(hiveConf, EVENT_ID, sourceTable, Arrays.asList(partition1), false, PARTITION_BASE_LOCATION, fileSystemFactory, sourceCatalogListener); assertThat(manager.getPartitionSubPath(new Path(partition1.getSd().getLocation())), is(new Path("partition1"))); }
@Test public void partitionSubPathUriEncoded() throws Exception { Path path = new Path(PARTITION_BASE_LOCATION + "/partition%251"); StorageDescriptor sd = new StorageDescriptor(); sd.setLocation(path.toUri().getPath()); partition1.setSd(sd); HdfsSnapshotLocationManager manager = new HdfsSnapshotLocationManager(hiveConf, EVENT_ID, sourceTable, Arrays.asList(partition1), false, PARTITION_BASE_LOCATION, fileSystemFactory, sourceCatalogListener); assertThat(manager.getPartitionSubPath(path), is(new Path("partition%251"))); }
@Test(expected = CircusTrainException.class) public void invalidPartitionSubPath() throws Exception { StorageDescriptor sd = new StorageDescriptor(); sd.setLocation("anotherBaseLocation" + "/partition1"); partition1.setSd(sd); HdfsSnapshotLocationManager manager = new HdfsSnapshotLocationManager(hiveConf, EVENT_ID, sourceTable, Arrays.asList(partition1), false, PARTITION_BASE_LOCATION, fileSystemFactory, sourceCatalogListener); manager.getPartitionSubPath(new Path(partition1.getSd().getLocation())); }