Move pytest-ethereum code to web3.tools

Author: njgheorghita

setup.py

@@ -76,6 +76,9 @@
         "requests>=2.16.0,<3.0.0",
         "websockets>=7.0.0,<8.0.0",
         "pypiwin32>=223;platform_system=='Windows'",
+        "ipfshttpclient>=0.4.12,<1",
+        "jsonschema>=2.6.0,<3",
+        "protobuf>=3.0.0,<4",
     ],
     setup_requires=['setuptools-markdown'],
     python_requires='>=3.6,<4',

tests/core/pm-module/conftest.py

@@ -17,10 +17,10 @@
 from ethpm.contract import (
     LinkableContract,
 )
-from pytest_ethereum import (
+from web3.tools import (
     linker,
 )
-from pytest_ethereum.deployer import (
+from web3.tools.pytest_ethereum.deployer import (
     Deployer,
 )
 

tests/ethpm/conftest.py

@@ -3,7 +3,7 @@
 
 from eth_utils.toolz import assoc_in
 import pytest
-from pytest_ethereum import linker as l  # noqa: E741
+from web3.tools import linker as l  # noqa: E741
 from web3 import Web3
 
 from ethpm import ASSETS_DIR, V2_PACKAGES_DIR, Package

web3/pm.py

@@ -31,25 +31,20 @@
     ASSETS_DIR,
     Package,
 )
-from ethpm.typing import (
+from eth_typing import (
     URI,
     Address,
     Manifest,
 )
-from ethpm.utils.backend import (
-    resolve_uri_contents,
-)
-from ethpm.utils.ipfs import (
-    is_ipfs_uri,
-)
-from ethpm.utils.manifest_validation import (
+from ethpm.validation.manifest import (
     validate_manifest_against_schema,
     validate_raw_manifest_format,
 )
-from ethpm.utils.uri import (
-    is_valid_content_addressed_github_uri,
+from ethpm.uri import (
+    is_supported_content_addressed_uri,
+    resolve_uri_contents,
 )
-from ethpm.validation import (
+from ethpm.validation.package import (
     validate_package_name,
     validate_package_version,
 )
@@ -659,7 +654,7 @@ def get_solidity_registry_manifest() -> Dict[str, Any]:
 
 
 def validate_is_supported_manifest_uri(uri):
-    if not is_ipfs_uri(uri) and not is_valid_content_addressed_github_uri(uri):
+    if not is_supported_content_addressed_uri(uri):
         raise ManifestValidationError(
             f"URI: {uri} is not a valid content-addressed URI. "
             "Currently only IPFS and Github content-addressed URIs are supported."

web3/tools/__init__.py

@@ -0,0 +1 @@
+from .pytest_ethereum import linker

web3/tools/pytest_ethereum/_utils.py

@@ -0,0 +1,116 @@
+from typing import Any, Dict, Iterable, List, Tuple
+
+from eth_typing import URI, Address, Manifest
+from eth_utils import to_canonical_address, to_dict, to_hex, to_list
+from eth_utils.toolz import assoc, assoc_in, dissoc
+from ethpm import Package
+from ethpm.uri import check_if_chain_matches_chain_uri
+from web3 import Web3
+
+from web3.tools.pytest_ethereum.exceptions import LinkerError
+
+
+def pluck_matching_uri(deployment_data: Dict[URI, Dict[str, str]], w3: Web3) -> URI:
+    """
+    Return any blockchain uri that matches w3-connected chain, if one
+    is present in the deployment data keys.
+    """
+    for uri in deployment_data.keys():
+        if check_if_chain_matches_chain_uri(w3, uri):
+            return uri
+    raise LinkerError(
+        f"No matching blockchain URI found in deployment_data: {list(deployment_data.keys())}, "
+        "for w3 instance: {w3.__repr__()}."
+    )
+
+
+def contains_matching_uri(deployment_data: Dict[str, Dict[str, str]], w3: Web3) -> bool:
+    """
+    Returns true if any blockchain uri in deployment data matches
+    w3-connected chain.
+    """
+    for uri in deployment_data.keys():
+        if check_if_chain_matches_chain_uri(w3, uri):
+            return True
+    return False
+
+
+def insert_deployment(
+    package: Package,
+    deployment_name: str,
+    deployment_data: Dict[str, str],
+    latest_block_uri: URI,
+) -> Manifest:
+    """
+    Returns a new manifest. If a matching chain uri is found in the old manifest, it will
+    update the chain uri along with the new deployment data. If no match, it will simply add
+    the new chain uri and deployment data.
+    """
+    old_deployments_data = package.manifest.get("deployments")
+    if old_deployments_data and contains_matching_uri(old_deployments_data, package.w3):
+        old_chain_uri = pluck_matching_uri(old_deployments_data, package.w3)
+        old_deployments_chain_data = old_deployments_data[old_chain_uri]
+        # Replace specific on-chain deployment (i.e. deployment_name)
+        new_deployments_chain_data_init = dissoc(
+            old_deployments_chain_data, deployment_name
+        )
+        new_deployments_chain_data = {
+            **new_deployments_chain_data_init,
+            **{deployment_name: deployment_data},
+        }
+        # Replace all on-chain deployments
+        new_deployments_data_init = dissoc(
+            old_deployments_data, "deployments", old_chain_uri
+        )
+        new_deployments_data = {
+            **new_deployments_data_init,
+            **{latest_block_uri: new_deployments_chain_data},
+        }
+        return assoc(package.manifest, "deployments", new_deployments_data)
+
+    return assoc_in(
+        package.manifest,
+        ("deployments", latest_block_uri, deployment_name),
+        deployment_data,
+    )
+
+
+@to_dict
+def create_deployment_data(
+    contract_name: str,
+    new_address: Address,
+    tx_receipt: Dict[str, Any],
+    link_refs: List[Dict[str, Any]] = None,
+) -> Iterable[Tuple[str, Any]]:
+    yield "contract_type", contract_name
+    yield "address", to_hex(new_address)
+    yield "transaction", to_hex(tx_receipt.transactionHash)
+    yield "block", to_hex(tx_receipt.blockHash)
+    if link_refs:
+        yield "runtime_bytecode", {"link_dependencies": create_link_dep(link_refs)}
+
+
+@to_list
+def create_link_dep(link_refs: List[Dict[str, Any]]) -> Iterable[Dict[str, Any]]:
+    for link_ref in link_refs:
+        yield {
+            "offsets": link_ref["offsets"],
+            "type": "reference",
+            "value": link_ref["name"],
+        }
+
+
+def get_deployment_address(linked_type: str, package: Package) -> Address:
+    """
+    Return the address of a linked_type found in a package's manifest deployments.
+    """
+    try:
+        deployment_address = to_canonical_address(
+            package.deployments.get(linked_type)["address"]
+        )
+    except KeyError:
+        raise LinkerError(
+            f"Package data does not contain a valid deployment of {linked_type} on the "
+            "current w3-connected chain."
+        )
+    return deployment_address

web3/tools/pytest_ethereum/deployer.py

@@ -0,0 +1,37 @@
+from typing import Any, Callable, Dict, Tuple  # noqa: F401
+
+from eth_typing import Address
+from ethpm import Package
+
+from web3.tools.pytest_ethereum.exceptions import DeployerError
+from web3.tools.pytest_ethereum.linker import deploy, linker
+
+
+class Deployer:
+    def __init__(self, package: Package) -> None:
+        if not isinstance(package, Package):
+            raise TypeError(
+                f"Expected a Package object, instead received {type(package)}."
+            )
+        self.package = package
+        self.strategies = {}  # type: Dict[str, Callable[[Package], Package]]
+
+    def deploy(
+        self, contract_type: str, *args: Any, **kwargs: Any
+    ) -> Tuple[Package, Address]:
+        factory = self.package.get_contract_factory(contract_type)
+        if contract_type in self.strategies:
+            strategy = self.strategies[contract_type]
+            return strategy(self.package)
+        if factory.needs_bytecode_linking:
+            raise DeployerError(
+                "Unable to deploy an unlinked factory. "
+                "Please register a strategy for this contract type."
+            )
+        strategy = linker(deploy(contract_type, *args, **kwargs))
+        return strategy(self.package)
+
+    def register_strategy(
+        self, contract_type: str, strategy: Callable[[Package], Package]
+    ) -> None:
+        self.strategies[contract_type] = strategy

web3/tools/pytest_ethereum/exceptions.py

@@ -0,0 +1,30 @@
+class PytestEthereumError(Exception):
+    """
+    Base class for all Pytest-Ethereum errors.
+    """
+
+    pass
+
+
+class DeployerError(PytestEthereumError):
+    """
+    Raised when the Deployer is unable to deploy a contract type.
+    """
+
+    pass
+
+
+class LinkerError(PytestEthereumError):
+    """
+    Raised when the Linker is unable to link two contract types.
+    """
+
+    pass
+
+
+class LogError(PytestEthereumError):
+    """
+    Raised when the Log class is instantiated with invalid arguments.
+    """
+
+    pass

web3/tools/pytest_ethereum/linker.py

@@ -0,0 +1,103 @@
+import logging
+from typing import Any, Callable, Dict, Tuple
+
+from eth_typing import Address
+from eth_utils import to_canonical_address, to_checksum_address, to_hex
+from eth_utils.toolz import assoc_in, curry, pipe
+from ethpm import Package
+from ethpm.uri import create_latest_block_uri
+
+from web3.tools.pytest_ethereum._utils import (
+    create_deployment_data,
+    get_deployment_address,
+    insert_deployment,
+)
+from web3.tools.pytest_ethereum.exceptions import LinkerError
+
+logger = logging.getLogger("pytest_ethereum.linker")
+
+
+def linker(*args: Callable[..., Any]) -> Callable[..., Any]:
+    return _linker(args)
+
+
+@curry
+def _linker(operations: Callable[..., Any], package: Package) -> Callable[..., Package]:
+    return pipe(package, *operations)
+
+
+def deploy(
+    contract_name: str, *args: Any, transaction: Dict[str, Any] = None
+) -> Callable[..., Tuple[Package, Address]]:
+    """
+    Return a newly created package and contract address.
+    Will deploy the given contract_name, if data exists in package. If
+    a deployment is found on the current w3 instance, it will return that deployment
+    rather than creating a new instance.
+    """
+    return _deploy(contract_name, args, transaction)
+
+
+@curry
+def _deploy(
+    contract_name: str, args: Any, transaction: Dict[str, Any], package: Package
+) -> Tuple[Package, Address]:
+    # Deploy new instance
+    factory = package.get_contract_factory(contract_name)
+    if not factory.linked_references and factory.unlinked_references:
+        raise LinkerError(
+            f"Contract factory: {contract_name} is missing runtime link references, which are "
+            "necessary to populate manifest deployments that have a link reference. If using the "
+            "builder tool, use `contract_type(..., runtime_bytecode=True)`."
+        )
+    tx_hash = factory.constructor(*args).transact(transaction)
+    tx_receipt = package.w3.eth.waitForTransactionReceipt(tx_hash)
+    address = to_canonical_address(tx_receipt.contractAddress)
+    # Create manifest copy with new deployment instance
+    latest_block_uri = create_latest_block_uri(package.w3, 0)
+    deployment_data = create_deployment_data(
+        contract_name, address, tx_receipt, factory.linked_references
+    )
+    manifest = insert_deployment(
+        package, contract_name, deployment_data, latest_block_uri
+    )
+    logger.info("%s deployed." % contract_name)
+    return Package(manifest, package.w3)
+
+
+@curry
+def link(contract: str, linked_type: str, package: Package) -> Package:
+    """
+    Return a new package, created with a new manifest after applying the linked type
+    reference to the contract factory.
+    """
+    deployment_address = get_deployment_address(linked_type, package)
+    unlinked_factory = package.get_contract_factory(contract)
+    if not unlinked_factory.needs_bytecode_linking:
+        raise LinkerError(
+            f"Contract factory: {unlinked_factory.__repr__()} does not need bytecode linking, "
+            "so it is not a valid contract type for link()"
+        )
+    linked_factory = unlinked_factory.link_bytecode({linked_type: deployment_address})
+    # todo replace runtime_bytecode in manifest
+    manifest = assoc_in(
+        package.manifest,
+        ("contract_types", contract, "deployment_bytecode", "bytecode"),
+        to_hex(linked_factory.bytecode),
+    )
+    logger.info(
+        "%s linked to %s at address %s."
+        % (contract, linked_type, to_checksum_address(deployment_address))
+    )
+    return Package(manifest, package.w3)
+
+
+@curry
+def run_python(callback_fn: Callable[..., None], package: Package) -> Package:
+    """
+    Return the unmodified package, after performing any user-defined callback function on
+    the contracts in the package.
+    """
+    callback_fn(package)
+    logger.info("%s python function ran." % callback_fn.__name__)
+    return package