Set up goth integration tests

Adds files necessary for running goth tests as part of yagna repository.
This also includes moving the VM E2E test from goth to yagna.

.github/workflows/rust.yml

@@ -104,12 +104,13 @@ jobs:
     name: Run integration tests
     runs-on: goth
     needs: build
+    defaults:
+      run:
+        working-directory: './goth_tests'
+
     steps:
       - name: Checkout
         uses: actions/checkout@v2
-        with:
-          repository: 'golemfactory/goth'
-          token: ${{ secrets.YAGNA_WORKFLOW_TOKEN }}
 
       - name: Configure python
         uses: actions/setup-python@v2
@@ -120,6 +121,7 @@ jobs:
         uses: Gr1N/setup-poetry@v4
         with:
           poetry-version: 1.1.4
+          working-directory: './goth_tests'
 
       - name: Install dependencies
         run: poetry install --no-root
@@ -140,7 +142,9 @@ jobs:
       - name: Run test suite
         env:
           GITHUB_API_TOKEN: ${{ secrets.GITHUB_TOKEN }}
-        run: poetry run poe ci_test_self_hosted --yagna-binary-path=/tmp/yagna-build
+        run: |
+          poetry run poe goth-assets
+          poetry run poe goth-tests --yagna-binary-path=/tmp/yagna-build
 
       - name: Upload test logs
         uses: actions/upload-artifact@v2

goth_tests/.gitignore

@@ -0,0 +1,4 @@
+.mypy_cache
+.pytest_cache
+__pycache__
+assets

goth_tests/__init__.py

@@ -0,0 +1 @@
+"""Root module for goth integration tests."""

goth_tests/conftest.py

@@ -0,0 +1,24 @@
+from datetime import datetime, timezone
+from pathlib import Path
+
+import pytest
+
+from goth.runner.log import configure_logging
+
+
+@pytest.fixture(scope="session")
+def common_assets() -> Path:
+    assets_path = Path(__file__).parent / "assets"
+    return assets_path.resolve()
+
+
+@pytest.fixture(scope="session")
+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)
+
+    configure_logging(log_dir)
+
+    return log_dir

goth_tests/domain/README.md

@@ -0,0 +1,3 @@
+## Domain-specific integration tests
+
+This directory tree includes test cases which focus on detailed aspects of yagna modules, rather than end-to-end flows.

goth_tests/domain/__init__.py

@@ -0,0 +1 @@
+"""Domain-specific integration tests."""

goth_tests/domain/exe_units/__init__.py

@@ -0,0 +1 @@
+"""Integration tests for yagna exe units."""

goth_tests/domain/market/__init__.py

@@ -0,0 +1 @@
+"""Integration tests for yagna market."""

goth_tests/domain/payments/__init__.py

@@ -0,0 +1 @@
+"""Integration tests for yagna payments."""

goth_tests/domain/ya-provider/__init__.py

@@ -0,0 +1 @@
+"""Integration tests for ya-provider."""

goth_tests/e2e/README.md

@@ -0,0 +1,3 @@
+## End-to-end integration tests
+
+This directory tree contains integration tests which run across the full stack of yagna components and include complete scenarios (both positive and negative).

goth_tests/e2e/__init__.py

@@ -0,0 +1 @@
+"""End-to-end integration tests.."""

goth_tests/e2e/vm/test_e2e_vm.py

@@ -0,0 +1,93 @@
+"""End to end tests for requesting VM tasks using goth REST API client."""
+
+import json
+import logging
+import os
+from pathlib import Path
+from typing import List
+
+import pytest
+
+from goth.address import (
+    PROXY_HOST,
+    YAGNA_REST_URL,
+)
+from goth.configuration import load_yaml
+from goth.node import node_environment
+from goth.runner import Runner
+from goth.runner.container.payment import PaymentIdPool
+from goth.runner.container.yagna import YagnaContainerConfig
+from goth.runner.probe import ProviderProbe, RequestorProbe
+
+from goth_tests.helpers.negotiation import DemandBuilder, negotiate_agreements
+from goth_tests.helpers.activity import vm_exe_script
+
+logger = logging.getLogger("goth.test.e2e_vm")
+
+
+@pytest.mark.asyncio
+async def test_e2e_vm_success(
+    common_assets: Path,
+    log_dir: Path,
+):
+    """Test successful flow requesting a Blender task with goth REST API client."""
+
+    goth_config = load_yaml(common_assets / "goth-config.yml")
+
+    runner = Runner(
+        base_log_dir=log_dir,
+        compose_config=goth_config.compose_config,
+        web_root_path=Path(__file__).parent / "assets",
+    )
+
+    async with runner(goth_config.containers):
+        task_package = (
+            "hash:sha3:9a3b5d67b0b27746283cb5f287c13eab1beaa12d92a9f536b747c7ae:"
+            "http://3.249.139.167:8000/local-image-c76719083b.gvmi"
+        )
+
+        output_file = "out0000.png"
+
+        output_path = Path(runner.web_root_path) / "upload" / output_file
+        if output_path.exists():
+            os.remove(output_path)
+
+        requestor = runner.get_probes(probe_type=RequestorProbe)[0]
+        providers = runner.get_probes(probe_type=ProviderProbe)
+
+        demand = DemandBuilder(requestor).props_from_template(task_package).build()
+
+        agreement_providers = await negotiate_agreements(
+            requestor,
+            demand,
+            providers,
+            lambda proposal: proposal.properties.get("golem.runtime.name") == "vm",
+        )
+
+        #  Activity
+        exe_script = vm_exe_script(runner, output_file)
+        num_commands = len(exe_script)
+
+        for agreement_id, provider in agreement_providers:
+            logger.info("Running activity on %s", provider.name)
+            activity_id = await requestor.create_activity(agreement_id)
+            await provider.wait_for_exeunit_started()
+            batch_id = await requestor.call_exec(activity_id, json.dumps(exe_script))
+            await requestor.collect_results(
+                activity_id, batch_id, num_commands, timeout=300
+            )
+            await requestor.destroy_activity(activity_id)
+            await provider.wait_for_exeunit_finished()
+
+        assert output_path.is_file()
+        assert output_path.stat().st_size > 0
+
+        # Payment
+
+        for agreement_id, provider in agreement_providers:
+            await provider.wait_for_invoice_sent()
+            invoices = await requestor.gather_invoices(agreement_id)
+            assert all(inv.agreement_id == agreement_id for inv in invoices)
+            # TODO:
+            await requestor.pay_invoices(invoices)
+            await provider.wait_for_invoice_paid()

goth_tests/helpers/__init__.py

@@ -0,0 +1 @@
+"""Helper functions and classes used in goth integration tests."""

goth_tests/helpers/activity.py

@@ -0,0 +1,75 @@
+"""Activity helpers."""
+
+import os
+from pathlib import Path
+
+from goth.runner import Runner
+
+
+def vm_exe_script(runner: Runner, output_file: str = "output.png"):
+    """VM exe script builder."""
+    """Create a VM exe script for running a Blender task."""
+
+    output_path = Path(runner.web_root_path) / output_file
+    if output_path.exists():
+        os.remove(output_path)
+
+    web_server_addr = f"http://{runner.host_address}:{runner.web_server_port}"
+
+    return [
+        {"deploy": {}},
+        {"start": {}},
+        {
+            "transfer": {
+                "from": f"{web_server_addr}/scene.blend",
+                "to": "container:/golem/resource/scene.blend",
+            }
+        },
+        {
+            "transfer": {
+                "from": f"{web_server_addr}/params.json",
+                "to": "container:/golem/work/params.json",
+            }
+        },
+        {"run": {"entry_point": "/golem/entrypoints/run-blender.sh", "args": []}},
+        {
+            "transfer": {
+                "from": f"container:/golem/output/{output_file}",
+                "to": f"{web_server_addr}/upload/{output_file}",
+            }
+        },
+    ]
+
+
+def wasi_exe_script(runner: Runner, output_file: str = "upload_file"):
+    """WASI exe script builder."""
+    """Create a WASI exe script for running a WASI tutorial task."""
+
+    output_path = Path(runner.web_root_path) / output_file
+    if output_path.exists():
+        os.remove(output_path)
+
+    web_server_addr = f"http://{runner.host_address}:{runner.web_server_port}"
+
+    return [
+        {"deploy": {}},
+        {"start": {"args": []}},
+        {
+            "transfer": {
+                "from": f"{web_server_addr}/params.json",
+                "to": "container:/input/file_in",
+            }
+        },
+        {
+            "run": {
+                "entry_point": "rust-wasi-tutorial",
+                "args": ["/input/file_in", "/output/file_cp"],
+            }
+        },
+        {
+            "transfer": {
+                "from": "container:/output/file_cp",
+                "to": f"{web_server_addr}/upload/{output_file}",
+            }
+        },
+    ]

goth_tests/helpers/negotiation.py

@@ -0,0 +1,115 @@
+"""Helper functions for building custom Offers and negotiating Agreements."""
+
+import logging
+from typing import List, Optional, Callable, Tuple, Any
+from datetime import datetime, timedelta
+
+from ya_market import Demand, DemandOfferBase, Proposal
+
+from goth.node import DEFAULT_SUBNET
+from goth.runner.probe import ProviderProbe, RequestorProbe
+
+
+logger = logging.getLogger(__name__)
+
+
+class DemandBuilder:
+    """Helper for building custom Demands.
+
+    Use if RequestorProbe.subscribe_template_demand function
+    is not enough for you.
+    """
+
+    def __init__(self, requestor: RequestorProbe):
+        self._requestor = requestor
+        self._properties = dict()
+        self._constraints = "()"
+        self._properties["golem.node.debug.subnet"] = DEFAULT_SUBNET
+
+    def props_from_template(self, task_package: str) -> "DemandBuilder":
+        """Build default properties."""
+
+        new_props = {
+            "golem.node.id.name": f"test-requestor-{self._requestor.name}",
+            "golem.srv.comp.expiration": int(
+                (datetime.now() + timedelta(minutes=10)).timestamp() * 1000
+            ),
+            "golem.srv.comp.task_package": task_package,
+        }
+        self._properties.update(new_props)
+        return self
+
+    def property(self, key: str, value: Any) -> "DemandBuilder":
+        """Add property."""
+        self._properties[key] = value
+        return self
+
+    # TODO: Building constraints.
+    def constraints(self, constraints: str) -> "DemandBuilder":
+        """Add constraints.
+
+        Note: This will override previous constraints.
+        """
+
+        self._constraints = constraints
+        return self
+
+    def build(self) -> DemandOfferBase:
+        """Create Demand from supplied parameters."""
+        return DemandOfferBase(
+            properties=self._properties,
+            constraints=self._constraints,
+        )
+
+
+async def negotiate_agreements(
+    requestor: RequestorProbe,
+    demand: Demand,
+    providers: List[ProviderProbe],
+    proposal_filter: Optional[Callable[[Proposal], bool]] = lambda p: True,
+) -> List[Tuple[str, ProviderProbe]]:
+    """Negotiate agreements with supplied providers.
+
+    Use negotiate_agreements function, when you don't need any custom negotiation
+    logic, but rather you want to test further parts of yagna protocol
+    and need ready Agreements.
+    """
+    for provider in providers:
+        await provider.wait_for_offer_subscribed()
+
+    subscription_id, demand = await requestor.subscribe_demand(demand)
+
+    proposals = await requestor.wait_for_proposals(
+        subscription_id,
+        providers,
+        proposal_filter,
+    )
+    logger.info("Collected %s proposals", len(proposals))
+
+    agreement_providers = []
+
+    for proposal in proposals:
+        provider = next(p for p in providers if p.address == proposal.issuer_id)
+        logger.info("Processing proposal from %s", provider.name)
+
+        counter_proposal_id = await requestor.counter_proposal(
+            subscription_id, demand, proposal
+        )
+        await provider.wait_for_proposal_accepted()
+
+        new_proposals = await requestor.wait_for_proposals(
+            subscription_id,
+            (provider,),
+            lambda proposal: proposal.prev_proposal_id == counter_proposal_id,
+        )
+
+        agreement_id = await requestor.create_agreement(new_proposals[0])
+        await requestor.confirm_agreement(agreement_id)
+        await provider.wait_for_agreement_approved()
+        await requestor.wait_for_approval(agreement_id)
+        agreement_providers.append((agreement_id, provider))
+
+    await requestor.unsubscribe_demand(subscription_id)
+    logger.info("Got %s agreements", len(agreement_providers))
+
+    return agreement_providers

goth_tests/helpers/payment.py

@@ -0,0 +1,19 @@
+"""Helper functions for easy handling of payments."""
+
+from typing import List, Tuple
+
+from goth.runner.probe import ProviderProbe, RequestorProbe
+
+
+async def pay_all(
+    requestor: RequestorProbe,
+    agreements: List[Tuple[str, ProviderProbe]],
+):
+    """Pay for all Agreements."""
+    for agreement_id, provider in agreements:
+        await provider.wait_for_invoice_sent()
+        invoices = await requestor.gather_invoices(agreement_id)
+        assert all(inv.agreement_id == agreement_id for inv in invoices)
+        # TODO:
+        await requestor.pay_invoices(invoices)
+        await provider.wait_for_invoice_paid()