Initial e2e test commit

.circleci/config.yml

@@ -19,12 +19,17 @@ jobs:
   test:
     machine:
       image: << pipeline.parameters.ubuntu_image >>
+    environment:
+      CI_E2E_FILENAME: "fa6ad40d/rel-nightly"
     steps:
       - go/install:
           version: "1.17.9"
+      - install_dependencies
       - build_conduit
       - install_linter
       - run_tests
+      # Change this to run_e2e_tests once we have stable algod binaries containing delta APIs
+      - run_e2e_tests_nightly
       - codecov/upload
 
 commands:
@@ -34,6 +39,20 @@ commands:
       - checkout
       - run: make
 
+  install_dependencies:
+    description: prepare machine for next steps
+    steps:
+      - checkout
+
+      - run:
+          name: Install python and other python dependencies
+          command: |
+            sudo apt update
+            sudo apt -y install python3 python3-pip python3-setuptools python3-wheel libboost-math-dev libffi-dev
+            pip3 install e2e_tests/
+
+      - run: echo 'export PATH=$PATH:/usr/local/go/bin:$HOME/.local/bin' >> $BASH_ENV
+
   install_linter:
     description: Install golangci-lint
     steps:
@@ -49,3 +68,23 @@ commands:
       - run:
           command: make test
           no_output_timeout: 15m
+
+  run_e2e_tests:
+    steps:
+      - run:
+          name: Install go-algorand stable binaries
+          command: |
+            wget https://raw.githubusercontent.com/algorand/go-algorand/rel/stable/cmd/updater/update.sh && chmod 744 update.sh
+            ./update.sh -i -c stable -n -d ./ -p /usr/local/go/bin
+            export GOPATH=/usr/local/go/
+      - run: make e2e-conduit
+
+  run_e2e_tests_nightly:
+    steps:
+      - run:
+          name: Install go-algorand nightly binaries
+          command: |
+            wget https://raw.githubusercontent.com/algorand/go-algorand/rel/stable/cmd/updater/update.sh && chmod 744 update.sh
+            ./update.sh -i -c nightly -n -d ./ -p /usr/local/go/bin
+            export GOPATH=/usr/local/go/
+      - run: make e2e-conduit

.gitignore

@@ -30,6 +30,7 @@ __pycache__
 e2e_tests/dist
 e2e_tests/build
 e2e_tests/*egg-info*
+e2e_tests/src/*egg-info*
 .venv
 
 # jetbrains IDE

Makefile

@@ -20,6 +20,10 @@ all: conduit
 conduit:
 	go generate ./... && cd cmd/conduit && go build -ldflags="${GOLDFLAGS}"
 
+# note: when running e2e tests manually be sure to set the e2e filename: 'export CI_E2E_FILENAME=rel-nightly'
+e2e-conduit: conduit
+	export PATH=$(GOPATH1)/bin:$(PATH); pip3 install e2e_tests/ && e2econduit --s3-source-net ${CI_E2E_FILENAME} --conduit-bin cmd/conduit/conduit
+
 # check that all packages (except tests) compile
 check:
 	go build ./...
@@ -34,4 +38,4 @@ lint:
 fmt:
 	go fmt ./...
 
-.PHONY: all conduit check test lint fmt
+.PHONY: all conduit check test lint fmt e2e-conduit

e2e_tests/pyproject.toml

@@ -0,0 +1,22 @@
+[build-system]
+requires = ["setuptools"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "conduit_e2e_tests"
+description = "End to end tests for Algorand Conduit"
+version = "0.0.1"
+requires-python = ">=3.8"
+dependencies = [
+    "boto3==1.24.71",
+    "msgpack==1.0.4",
+    "py-algorand-sdk==1.17.0",
+    "pytest==6.2.5",
+    "PyYAML==6.0",
+    "setuptools ==65.3.0",
+]
+
+
+[project.scripts]
+e2econduit = "e2e_conduit.e2econduit:main"
+get-test-data = "e2e_common.get_test_data:main"

e2e_tests/src/e2e_common/util.py

@@ -0,0 +1,203 @@
+#!/usr/bin/env python3
+
+import atexit
+import logging
+import os
+import random
+import sqlite3
+import subprocess
+import sys
+import time
+
+import msgpack
+
+logger = logging.getLogger(__name__)
+
+
+def maybedecode(x):
+    if hasattr(x, "decode"):
+        return x.decode()
+    return x
+
+
+def mloads(x):
+    return msgpack.loads(x, strict_map_key=False, raw=True)
+
+
+def unmsgpack(ob):
+    "convert dict from msgpack.loads() with byte string keys to text string keys"
+    if isinstance(ob, dict):
+        od = {}
+        for k, v in ob.items():
+            k = maybedecode(k)
+            okv = False
+            if (not okv) and (k == "note"):
+                try:
+                    v = unmsgpack(mloads(v))
+                    okv = True
+                except:
+                    pass
+            if (not okv) and k in ("type", "note"):
+                try:
+                    v = v.decode()
+                    okv = True
+                except:
+                    pass
+            if not okv:
+                v = unmsgpack(v)
+            od[k] = v
+        return od
+    if isinstance(ob, list):
+        return [unmsgpack(v) for v in ob]
+    # if isinstance(ob, bytes):
+    #    return base64.b64encode(ob).decode()
+    return ob
+
+
+def _getio(p, od, ed):
+    if od is not None:
+        od = maybedecode(od)
+    elif p.stdout:
+        try:
+            od = maybedecode(p.stdout.read())
+        except:
+            logger.error("subcomand out", exc_info=True)
+    if ed is not None:
+        ed = maybedecode(ed)
+    elif p.stderr:
+        try:
+            ed = maybedecode(p.stderr.read())
+        except:
+            logger.error("subcomand err", exc_info=True)
+    return od, ed
+
+
+def xrun(cmd, *args, **kwargs):
+    timeout = kwargs.pop("timeout", None)
+    kwargs["stdout"] = subprocess.PIPE
+    kwargs["stderr"] = subprocess.STDOUT
+    cmdr = " ".join(map(repr, cmd))
+    try:
+        p = subprocess.Popen(cmd, *args, **kwargs)
+    except Exception as e:
+        logger.error("subprocess failed {}".format(cmdr), exc_info=True)
+        raise
+    stdout_data, stderr_data = None, None
+    try:
+        if timeout:
+            stdout_data, stderr_data = p.communicate(timeout=timeout)
+        else:
+            stdout_data, stderr_data = p.communicate()
+    except subprocess.TimeoutExpired as te:
+        logger.error("subprocess timed out {}".format(cmdr), exc_info=True)
+        stdout_data, stderr_data = _getio(p, stdout_data, stderr_data)
+        if stdout_data:
+            sys.stderr.write("output from {}:\n{}\n\n".format(cmdr, stdout_data))
+        if stderr_data:
+            sys.stderr.write("stderr from {}:\n{}\n\n".format(cmdr, stderr_data))
+        raise
+    except Exception as e:
+        logger.error("subprocess exception {}".format(cmdr), exc_info=True)
+        stdout_data, stderr_data = _getio(p, stdout_data, stderr_data)
+        if stdout_data:
+            sys.stderr.write("output from {}:\n{}\n\n".format(cmdr, stdout_data))
+        if stderr_data:
+            sys.stderr.write("stderr from {}:\n{}\n\n".format(cmdr, stderr_data))
+        raise
+    if p.returncode != 0:
+        logger.error("cmd failed ({}) {}".format(p.returncode, cmdr))
+        stdout_data, stderr_data = _getio(p, stdout_data, stderr_data)
+        if stdout_data:
+            sys.stderr.write("output from {}:\n{}\n\n".format(cmdr, stdout_data))
+        if stderr_data:
+            sys.stderr.write("stderr from {}:\n{}\n\n".format(cmdr, stderr_data))
+        raise Exception("error: cmd failed: {}".format(cmdr))
+    if logger.isEnabledFor(logging.DEBUG):
+        logger.debug(
+            "cmd success: %s\n%s\n%s\n",
+            cmdr,
+            maybedecode(stdout_data),
+            maybedecode(stderr_data),
+        )
+
+
+def atexitrun(cmd, *args, **kwargs):
+    cargs = [cmd] + list(args)
+    atexit.register(xrun, *cargs, **kwargs)
+
+
+def find_binary(binary, exc=True, binary_name="algorand-indexer"):
+    if binary:
+        return binary
+    # manually search local build and PATH for binary_name
+    path = [f"cmd/{binary_name}"] + os.getenv("PATH").split(":")
+    for pd in path:
+        ib = os.path.join(pd, binary_name)
+        if os.path.exists(ib):
+            return ib
+    msg = f"could not find {binary_name} at the provided location or PATH environment variable."
+    if exc:
+        raise Exception(msg)
+    logger.error(msg)
+    return None
+
+
+def ensure_test_db(connection_string, keep_temps=False):
+    if connection_string:
+        # use the passed db
+        return connection_string
+    # create a temporary database
+    dbname = "e2eindex_{}_{}".format(int(time.time()), random.randrange(1000))
+    xrun(["dropdb", "--if-exists", dbname], timeout=5)
+    xrun(["createdb", dbname], timeout=5)
+    if not keep_temps:
+        atexitrun(["dropdb", "--if-exists", dbname], timeout=5)
+    else:
+        logger.info("leaving db %r", dbname)
+    return "dbname={} sslmode=disable".format(dbname)
+
+
+# whoever calls this will need to import boto and get the s3 client
+def firstFromS3Prefix(
+    s3, bucket, prefix, desired_filename, outdir=None, outpath=None
+) -> bool:
+    haystack = []
+    found_needle = False
+
+    response = s3.list_objects_v2(Bucket=bucket, Prefix=prefix, MaxKeys=100)
+    if (not response.get("KeyCount")) or ("Contents" not in response):
+        raise Exception("nothing found in s3://{}/{}".format(bucket, prefix))
+    for x in response["Contents"]:
+        path = x["Key"]
+        haystack.append(path)
+        _, fname = path.rsplit("/", 1)
+        if fname == desired_filename:
+            if outpath is None:
+                if outdir is None:
+                    outdir = "."
+                outpath = os.path.join(outdir, desired_filename)
+            logger.info("s3://%s/%s -> %s", bucket, x["Key"], outpath)
+            s3.download_file(bucket, x["Key"], outpath)
+            found_needle = True
+            break
+
+    if not found_needle:
+        logger.warning("file {} not found in s3://{}/{}".format(desired_filename, bucket, prefix))
+    return found_needle
+
+
+def countblocks(path):
+    db = sqlite3.connect(path)
+    cursor = db.cursor()
+    cursor.execute("SELECT max(rnd) FROM blocks")
+    row = cursor.fetchone()
+    cursor.close()
+    db.close()
+    return row[0]
+
+
+def hassuffix(x, *suffixes):
+    for s in suffixes:
+        if x.endswith(s):
+            return True
+    return False

e2e_tests/src/e2e_conduit/e2econduit.py

@@ -0,0 +1,67 @@
+#!/usr/bin/env python3
+#
+
+import argparse
+import logging
+import os
+import sys
+
+from e2e_common.util import find_binary
+import e2e_conduit.fixtures.importers as importers
+import e2e_conduit.fixtures.processors as processors
+import e2e_conduit.fixtures.exporters as exporters
+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__)
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    # TODO FIXME convert keep_temps to debug mode which will leave all resources running/around
+    # So files will not be deleted and docker containers will be left running
+    ap.add_argument("--keep-temps", default=False, action="store_true")
+    ap.add_argument(
+        "--conduit-bin",
+        default=None,
+        help="path to conduit binary, otherwise search PATH",
+    )
+    ap.add_argument(
+        "--source-net",
+        help="Path to test network directory containing Primary and other nodes. May be a tar file.",
+    )
+    ap.add_argument(
+        "--s3-source-net",
+        help="AWS S3 key suffix to test network tarball containing Primary and other nodes. Must be a tar bz2 file.",
+    )
+    ap.add_argument("--verbose", default=False, action="store_true")
+    args = ap.parse_args()
+    if args.verbose:
+        logging.basicConfig(level=logging.DEBUG)
+    else:
+        logging.basicConfig(level=logging.INFO)
+    sourcenet = args.source_net
+    source_is_tar = False
+    if not sourcenet:
+        e2edata = os.getenv("E2EDATA")
+        sourcenet = e2edata and os.path.join(e2edata, "net")
+    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)
+
+    success = True
+    for scenario in scenarios:
+        runner.setup_scenario(scenario)
+        if runner.run_scenario(scenario) != 0:
+            success = False
+    return 0 if success else 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())

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

@@ -0,0 +1 @@
+from e2e_conduit.fixtures.exporters.postgresql import PostgresqlExporter