Add integration test that starts blender example in goth

.github/workflows/integration.yml

@@ -0,0 +1,66 @@
+name: integration-test
+
+on:
+  push:
+    branches:
+      - master
+      # - <your-branch>    # put your branch name here to test it @ GH Actions
+  pull_request:
+    branches:
+      - master
+      - b0.*
+
+jobs:
+
+  goth-tests:
+    name: Run integration tests
+    runs-on: goth
+    steps:
+
+      - name: Checkout
+        uses: actions/checkout@v2
+
+      - name: Configure python
+        uses: actions/setup-python@v2
+        with:
+          python-version: '3.8.0'
+
+      - name: Configure poetry
+        uses: Gr1N/setup-poetry@v4
+        with:
+          poetry-version: 1.1.4
+
+      - name: Install dependencies
+        run: |
+          poetry env use python3.8
+          poetry install -E integration-tests
+
+      - name: Log in to GitHub Docker repository
+        run: echo ${{ secrets.GITHUB_TOKEN }} | docker login docker.pkg.github.com -u ${{github.actor}} --password-stdin
+
+      - name: Run test suite
+        env:
+          GITHUB_API_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+        run: |
+          poetry run poe goth-assets
+          poetry run poe goth-tests
+
+      - name: Upload test logs
+        uses: actions/upload-artifact@v2
+        if: always()
+        with:
+           name: goth-logs
+           path: /tmp/goth-tests
+
+      # Only relevant for self-hosted runners
+      - name: Remove test logs
+        if: always()
+        run: rm -rf /tmp/goth-tests
+
+      # Only relevant for self-hosted runners
+      - name: Remove poetry virtual env
+        if: always()
+        # Python version below should agree with the version set up by this job.
+        # In future we'll be able to use the `--all` flag here to remove envs for
+        # all Python versions (https://github.com/python-poetry/poetry/issues/3208).
+        run: poetry env remove python3.8

pyproject.toml

@@ -1,6 +1,6 @@
 [build-system]
-requires = ["poetry"]
-build-backend = "poetry.masonry.api"
+requires = ["poetry_core>=1.0.0"]
+build-backend = "poetry.core.masonry.api"
 
 [tool.poetry]
 name = "yapapi"
@@ -37,10 +37,18 @@ toml = "^0.10.1"
 srvresolver = "^0.3.5"
 colorama = "^0.4.4"
 
+# Adding `goth` to dependencies causes > 40 additional packages to be installed. Given
+# that dependency resolution in `poetry` is rather slow, we'd like to avoid installing
+# `goth` for example in a CI pipeline that only runs linters/unit tests, not integration
+# tests. Therefore we specify `goth` as an "extra" dependency, with `optional = "true"`.
+# It will be then installable with `poetry install -E integration-tests`.
+# Note that putting `goth` in `poetry.dev-dependencies` instead of `poetry.dependencies`
+# would not work: see https://github.com/python-poetry/poetry/issues/129.
+goth = {git = "https://github.com/golemfactory/goth.git", branch = "master", optional = true, python = "^3.8.0"}
 
 [tool.poetry.extras]
 cli = ['fire', 'rich']
-
+integration-tests = ['goth', 'pytest', 'pytest-asyncio']
 
 [tool.poetry.dev-dependencies]
 black = "^20.8b1"
@@ -95,7 +103,9 @@ ya-client-payment = "0.1.0"
 ya-market = "0.1.0"
 
 [tool.poe.tasks]
-test = "pytest --cov=yapapi"
+test = "pytest --cov=yapapi --ignore tests/goth"
+goth-assets = "python -m goth create-assets tests/goth/assets"
+goth-tests = "pytest -svx tests/goth"
 typecheck = "mypy ."
 codestyle = "black --check --diff ."
 _liccheck_export = "poetry export -E cli -f requirements.txt -o .requirements.txt"

tests/goth/conftest.py

@@ -0,0 +1,19 @@
+from datetime import datetime, timezone
+from pathlib import Path
+
+import pytest
+
+
+@pytest.fixture(scope="session")
+def project_dir() -> Path:
+    package_dir = Path(__file__).parent.parent
+    return package_dir.parent.resolve()
+
+
+@pytest.fixture()
+def log_dir() -> Path:
+    base_dir = Path("/", "tmp", "goth-tests")
+    date_str = datetime.now(tz=timezone.utc).strftime("%Y%m%d_%H%M%S%z")
+    log_dir = base_dir / f"goth_{date_str}"
+    log_dir.mkdir(parents=True)
+    return log_dir

tests/goth/test_run_blender.py

@@ -0,0 +1,127 @@
+import logging
+import os
+from pathlib import Path
+import re
+
+import pytest
+
+from goth.assertions import EventStream
+from goth.configuration import load_yaml
+from goth.runner.log import configure_logging
+from goth.runner import Runner
+from goth.runner.probe import RequestorProbe
+
+
+logger = logging.getLogger("goth.test.run_blender")
+
+ALL_TASKS = {0, 10, 20, 30, 40, 50}
+
+
+# Temporal assertions expressing properties of sequences of "events". In this case, each "event"
+# is just a line of output from `blender.py`.
+
+
+async def assert_no_errors(output_lines: EventStream[str]):
+    """Assert that no output line contains the substring `ERROR`."""
+    async for line in output_lines:
+        if "ERROR" in line:
+            raise AssertionError("Command reported ERROR")
+
+
+async def assert_all_tasks_processed(status: str, output_lines: EventStream[str]):
+    """Assert that for every task in `ALL_TASKS` a line with `Task {status}` will appear."""
+    remaining_tasks = ALL_TASKS.copy()
+
+    async for line in output_lines:
+        m = re.search(rf".*Task {status} .* task data: ([0-9]+)", line)
+        if m:
+            task_data = int(m.group(1))
+            logger.debug("assert_all_tasks_processed: Task %s: %d", status, task_data)
+            remaining_tasks.discard(task_data)
+        if not remaining_tasks:
+            return
+
+    raise AssertionError(f"Tasks not {status}: {remaining_tasks}")
+
+
+async def assert_all_tasks_sent(output_lines: EventStream[str]):
+    """Assert that for every task a line with `Task sent` will appear."""
+    await assert_all_tasks_processed("sent", output_lines)
+
+
+async def assert_all_tasks_computed(output_lines: EventStream[str]):
+    """Assert that for every task a line with `Task computed` will appear."""
+    await assert_all_tasks_processed("computed", output_lines)
+
+
+async def assert_all_invoices_accepted(output_lines: EventStream[str]):
+    """Assert that an invoice is accepted for every provider that confirmed an agreement."""
+    unpaid_agreement_providers = set()
+
+    async for line in output_lines:
+        m = re.search("Agreement confirmed by provider '([^']*)'", line)
+        if m:
+            prov_name = m.group(1)
+            logger.debug("assert_all_invoices_accepted: adding provider '%s'", prov_name)
+            unpaid_agreement_providers.add(prov_name)
+        m = re.search("Accepted invoice from '([^']*)'", line)
+        if m:
+            prov_name = m.group(1)
+            logger.debug("assert_all_invoices_accepted: adding invoice for '%s'", prov_name)
+            unpaid_agreement_providers.remove(prov_name)
+
+    if unpaid_agreement_providers:
+        raise AssertionError(f"Unpaid agreements for: {','.join(unpaid_agreement_providers)}")
+
+
+@pytest.mark.asyncio
+async def test_run_blender(
+    log_dir: Path,
+    project_dir: Path,
+) -> None:
+
+    # This is the default configuration with 2 wasm/VM providers
+    goth_config = load_yaml(Path(__file__).parent / "assets" / "goth-config.yml")
+
+    blender_path = project_dir / "examples" / "blender" / "blender.py"
+
+    configure_logging(log_dir)
+
+    runner = Runner(
+        base_log_dir=log_dir,
+        compose_config=goth_config.compose_config,
+    )
+
+    async with runner(goth_config.containers):
+
+        requestor = runner.get_probes(probe_type=RequestorProbe)[0]
+
+        async with requestor.run_command_on_host(
+            f"{blender_path} --subnet-tag goth",
+            env=os.environ,
+        ) as (_cmd_task, cmd_monitor):
+
+            # Add assertions to the command output monitor `cmd_monitor`:
+            cmd_monitor.add_assertion(assert_no_errors)
+            cmd_monitor.add_assertion(assert_all_invoices_accepted)
+            all_sent = cmd_monitor.add_assertion(assert_all_tasks_sent)
+            all_computed = cmd_monitor.add_assertion(assert_all_tasks_computed)
+
+            await cmd_monitor.wait_for_pattern(".*Received proposals from 2 ", timeout=10)
+            logger.info("Received proposals")
+
+            await cmd_monitor.wait_for_pattern(".*Agreement proposed ", timeout=10)
+            logger.info("Agreement proposed")
+
+            await cmd_monitor.wait_for_pattern(".*Agreement confirmed ", timeout=10)
+            logger.info("Agreement confirmed")
+
+            await all_sent.wait_for_result(timeout=120)
+            logger.info("All tasks sent")
+
+            await all_computed.wait_for_result(timeout=30)
+            logger.info("All tasks computed, waiting for Executor shutdown")
+
+            await cmd_monitor.wait_for_pattern(".*Executor has shut down", timeout=120)
+
+            logger.info("Requestor script finished")