e2e-tests: Filter processor e2e test (#1493)

e2e_tests/README.md

@@ -63,3 +63,53 @@ make e2e
 ```
 
 Alternatively, modify `e2e_tests/docker/indexer/Dockerfile` by swapping out `$CI_E2E_FILENAME` with the test artifact you want to use. See `.circleci/config.yml` for the current values used by CI.
+
+### Creating new Conduit e2e tests
+
+## Creating a new plugin fixture
+All plugins for e2e tests are organized in `e2e_tests/src/e2e_conduit/fixtures/` under the proper directory structure. For example, processor plugin fixtures are under `e2e_tests/src/e2e_conduit/fixtures/processors/`.
+
+To create a new plugin, create a new class under the proper directory which subclasses `PluginFixture`.  
+__Note that some fixtures have further subclasses, such as ImporterPlugin, which support additional required behavior for those plugin types.__
+
+### Main Methods of PluginFixtures
+
+* `name`  
+Returns the name of the plugin--must be equivalent to the name which would be used in the conduit config.
+
+* `setup(self, accumulated_config)`  
+Setup is run before any of the config data is resolved. It accepts an `accumulated_config` which is a map containing all of the previously output config values.
+If your plugin needs to be fed some data, such as the algod directory which was created by the importer, this data can be retrieved from the accumulated_config.
+
+The setup method is responsible for any preparation your plugin needs to do before the pipeline starts.
+
+* `resolve_config_input`  
+This method sets values on `self.config_input`, a map which contains all of the Conduit config required to run your plugin. Any values set on this map will be serialized into
+the data section of your plugin's config when running it in Conduit.
+
+* `resolve_config_output`  
+Here, similarly to `config_input`, we set values on the `config_output` map. This map is what will be passed between plugins during initialization via the `accumulated_config`. If your plugin is creating any resources or initializing any values which other plugins need to know about, they should be set in the `config_output`.
+
+
+## Creating a new scenario using your plugin
+A `Scenario` is an abstraction for a given instantiation of a Conduit pipeline. In order to run a coduit e2e test, create a scenario in `e2e_tests/src/e2e_conduit/scenarios` and ensure that it is run in `e2e_tests/src/e2e_conduit/e2econduit.py`.
+
+For example,
+
+```
+Scenario(
+        "app_filter_indexer_scenario",
+        importer=importers.FollowerAlgodImporter(sourcenet),
+        processors=[
+            processors.FilterProcessor([
+                {"any": [
+                    {"tag": "txn.type",
+                     "expression-type": "equal",
+                     "expression": "appl"
+                    }
+                ]}
+            ]),
+        ],
+        exporter=exporters.PostgresqlExporter(),
+    )
+```

e2e_tests/src/e2e_conduit/e2econduit.py

@@ -13,6 +13,7 @@
 from e2e_conduit.runner import ConduitE2ETestRunner
 from e2e_conduit.scenarios import scenarios
 from e2e_conduit.scenarios.follower_indexer_scenario import follower_indexer_scenario
+from e2e_conduit.scenarios.filter_scenario import app_filter_indexer_scenario, pay_filter_indexer_scenario
 
 logger = logging.getLogger(__name__)
 
@@ -49,6 +50,8 @@ def main():
     importer_source = sourcenet if sourcenet else args.s3_source_net
     if importer_source:
         scenarios.append(follower_indexer_scenario(importer_source))
+        scenarios.append(app_filter_indexer_scenario(importer_source))
+        scenarios.append(pay_filter_indexer_scenario(importer_source))
 
     runner = ConduitE2ETestRunner(args.conduit_bin, keep_temps=args.keep_temps)
 

e2e_tests/src/e2e_conduit/fixtures/processors/__init__.py

@@ -0,0 +1 @@
+from e2e_conduit.fixtures.processors.filter import FilterProcessor

e2e_tests/src/e2e_conduit/fixtures/processors/filter.py

@@ -0,0 +1,25 @@
+from e2e_conduit.fixtures.plugin_fixture import PluginFixture
+
+
+''' FilterProcessor is a simple passthrough for the `filter_processor`
+    which sets the scenario's filters to the initialized value
+'''
+class FilterProcessor(PluginFixture):
+    def __init__(self, filters, search_inner=True):
+        self.filters = filters
+        self.search_inner = search_inner
+        super().__init__()
+
+    @property
+    def name(self):
+        return "filter_processor"
+
+    def resolve_config_input(self):
+        self.config_input["filters"] = self.filters
+        self.config_input["search-inner"] = self.search_inner
+
+    def resolve_config_output(self):
+        pass
+
+    def setup(self, accumulated_config):
+        pass

e2e_tests/src/e2e_conduit/scenarios/filter_scenario.py

@@ -0,0 +1,40 @@
+import e2e_conduit.fixtures.importers as importers
+import e2e_conduit.fixtures.processors as processors
+import e2e_conduit.fixtures.exporters as exporters
+from e2e_conduit.scenarios import Scenario, scenarios
+
+
+def app_filter_indexer_scenario(sourcenet):
+    return Scenario(
+        "app_filter_indexer_scenario",
+        importer=importers.FollowerAlgodImporter(sourcenet),
+        processors=[
+            processors.FilterProcessor([
+                {"any": [
+                    {"tag": "txn.type",
+                     "expression-type": "equal",
+                     "expression": "appl"
+                    }
+                ]}
+            ]),
+        ],
+        exporter=exporters.PostgresqlExporter(),
+    )
+
+
+def pay_filter_indexer_scenario(sourcenet):
+    return Scenario(
+        "pay_filter_indexer_scenario",
+        importer=importers.FollowerAlgodImporter(sourcenet),
+        processors=[
+            processors.FilterProcessor([
+                {"any": [
+                    {"tag": "txn.type",
+                     "expression-type": "equal",
+                     "expression": "pay"
+                    }
+                ]}
+            ]),
+        ],
+        exporter=exporters.PostgresqlExporter(),
+    )

e2e_tests/src/e2e_indexer/e2elive.py

@@ -1,6 +1,4 @@
 #!/usr/bin/env python3
-#
-
 import atexit
 import glob
 import gzip