### Question:
RenameFilterPlugin implements FilterPlugin { @Override public void transaction(ConfigSource config, Schema inputSchema, FilterPlugin.Control control) { PluginTask task = config.loadConfig(PluginTask.class); Map<String, String> renameMap = task.getRenameMap(); List<ConfigSource> rulesList = task.getRulesList(); for (String columnName : renameMap.keySet()) { inputSchema.lookupColumn(columnName); } Schema.Builder builder = Schema.builder(); for (Column column : inputSchema.getColumns()) { String name = column.getName(); if (renameMap.containsKey(name)) { name = renameMap.get(name); } builder.add(name, column.getType()); } Schema intermediateSchema = builder.build(); Schema outputSchema = intermediateSchema; for (ConfigSource rule : rulesList) { outputSchema = applyRule(rule, intermediateSchema); intermediateSchema = outputSchema; } control.run(task.dump(), outputSchema); }  @Override void transaction(ConfigSource config, Schema inputSchema,
                            FilterPlugin.Control control); @Override PageOutput open(TaskSource taskSource, Schema inputSchema,
                           Schema outputSchema, PageOutput output);  }### Answer:
@Test public void throwSchemaConfigExceptionIfColumnNotFound() { ConfigSource pluginConfig = Exec.newConfigSource() .set("columns", ImmutableMap.of("not_found", "any_name")); try { filter.transaction(pluginConfig, SCHEMA, new FilterPlugin.Control() { public void run(TaskSource task, Schema schema) {} }); fail(); } catch (Throwable t) { assertTrue(t instanceof SchemaConfigException); } }
@Test public void checkRenaming() { ConfigSource pluginConfig = Exec.newConfigSource() .set("columns", ImmutableMap.of("_c0", "_c0_new")); filter.transaction(pluginConfig, SCHEMA, new FilterPlugin.Control() { @Override public void run(TaskSource task, Schema newSchema) { Column old0 = SCHEMA.getColumn(0); Column new0 = newSchema.getColumn(0); assertEquals("_c0_new", new0.getName()); assertEquals(old0.getType(), new0.getType()); Column old1 = SCHEMA.getColumn(1); Column new1 = newSchema.getColumn(1); assertEquals("_c1", new1.getName()); assertEquals(old1.getType(), new1.getType()); } }); }
@Test public void checkConfigExceptionIfUnknownStringTypeOfRenamingOperator() { ConfigSource pluginConfig = Exec.newConfigSource() .set("rules", ImmutableList.of("string_rule")); try { filter.transaction(pluginConfig, SCHEMA, new FilterPlugin.Control() { public void run(TaskSource task, Schema schema) {} }); fail(); } catch (Throwable t) { assertTrue(t instanceof ConfigException); } }
@Test public void checkConfigExceptionIfUnknownListTypeOfRenamingOperator() { ConfigSource pluginConfig = Exec.newConfigSource() .set("rules", ImmutableList.of(ImmutableList.of("listed_operator1", "listed_operator2"))); try { filter.transaction(pluginConfig, SCHEMA, new FilterPlugin.Control() { public void run(TaskSource task, Schema schema) {} }); fail(); } catch (Throwable t) { assertTrue(t instanceof ConfigException); } }
@Test public void checkConfigExceptionIfUnknownRenamingOperatorName() { ConfigSource pluginConfig = Exec.newConfigSource() .set("rules", ImmutableList.of(ImmutableMap.of("rule", "some_unknown_renaming_operator"))); try { filter.transaction(pluginConfig, SCHEMA, new FilterPlugin.Control() { public void run(TaskSource task, Schema schema) {} }); fail(); } catch (Throwable t) { assertTrue(t instanceof ConfigException); } }