Week 06

.github/workflows/test.yaml

@@ -18,7 +18,7 @@ jobs:
       - name: Set up python
         uses: actions/setup-python@v4
         with:
-          python-version: '3.8'
+          python-version: '3.11'
       #----------------------------------------------
       #  -----  install & configure poetry  -----
       #----------------------------------------------
@@ -27,7 +27,7 @@ jobs:
         uses: actions/cache@v3
         with:
           path: ~/.local  # the path depends on the OS
-          key: poetry-1  # increment to reset cache
+          key: poetry-3  # increment to reset cache
       - name: Install Poetry
         if: steps.cached-poetry.outputs.cache-hit != 'true'
         uses: snok/install-poetry@v1
@@ -49,12 +49,12 @@ jobs:
       #----------------------------------------------
       - name: Install dependencies
         if: steps.cached-poetry-dependencies.outputs.cache-hit != 'true'
-        run: poetry install --no-interaction --no-root
+        run: poetry install --all-extras --no-interaction --no-root
       #----------------------------------------------
       # install your root project, if required
       #----------------------------------------------
       - name: Install project
-        run: poetry install --no-interaction
+        run: poetry install --all-extras --no-interaction
       #----------------------------------------------
       #                run linter
       #----------------------------------------------
@@ -66,6 +66,6 @@ jobs:
       #----------------------------------------------
       - name: Run pytest
         run: |
-          poetry run pytest
+          poetry run pytest -n auto
         env:
           TEST_SOLVED: 1

README.md

@@ -25,7 +25,7 @@ Join the opshin [discord server](https://discord.com/invite/umR3A2g4uw) for Q/A
 
 ## Installation
 
-1. Install Python 3.8, 3.9, 3.10 or 3.11 (if it not already installed on your operating system).
+1. Install Python 3.9, 3.10 or 3.11 (if it not already installed on your operating system).
 Python3.11 Installer [download](https://www.python.org/downloads/release/python-3112/).
 
 2. Install python poetry.
@@ -169,11 +169,22 @@ We use pycardano, but you can compare and contrast alternatives.
 ### [Lecture 6](https://www.youtube.com/playlist?list=PLNEK_Ejlx3x08fHgl_ZTlowVO8bjqITEh)
 
 - [The State Monad in practice](https://www.youtube.com/watch?v=8tWzG0ML6Z4&list=PLNEK_Ejlx3x08fHgl_ZTlowVO8bjqITEh&index=1)
+  - You can skip this for opshin.
 - [Introduction to the Plutus Simple Model library](https://youtu.be/Sft02LeXA_U)
+  - We implement `MockChainContext` and `MockUser` in `src/utils/mock.py`.
+    These classes allow us to easily test and evaluate our opshin contracts without the Cardano Node!
+  - We implement a simple test in `src/week06/tests/test_mock.py` with simulated spending and multiple users.
 - [Unit Testing a Smart Contract](https://youtu.be/vB8hyVq3HVo)
+  - Unit tests located in `src/week06/tests/test_negative_r_timed.py`
 - [Property Testing a Smart Contract](https://youtu.be/pF8HpKmaQi4)
+  - Property tests also located in `src/week06/tests/test_negative_r_timed.py`
+  - Read the documentation on [hypothesis](https://hypothesis.readthedocs.io/en/latest/)
+    to get familiar with property testing in Python.
 - [Testing Smart Contracts with Lucid](https://youtu.be/aUrIuDQgg5c)
+  - N/A.
 - [Double Spending and Homework](https://youtu.be/AZVpkwRhEaY)
+  - Complete the following test `src/week06/homework/test_exploitable_swap.py`
+  - Use your completed test to implement a fix to the swap script: `src/week06/homework/fixed_swap.py`
 
 
 ### [Lecture 7](https://www.youtube.com/playlist?list=PLNEK_Ejlx3x0wH_y1lQp4xtrkuaYSWi6V)

pyproject.toml

@@ -7,17 +7,22 @@ readme = "README.md"
 packages = [{include = "src"}]
 
 [tool.poetry.dependencies]
-python = ">=3.8,<3.12"
-opshin = "^0.12.5"
-pycardano = "^0.8.0"
+python = ">=3.9,<3.12"
+opshin = "^0.13.0"
+pycardano = {git = "https://github.com/juliusfrost/pycardano.git", rev = "d873e56e0aed2bcd92964f19928254510573fe5f"}
 click = "^8.1.3"
+pyaiken = {version = "^0.5.0", optional = true}
 
 
 [tool.poetry.group.dev.dependencies]
 black = "^23.3.0"
 pre-commit = "^3.2.2"
 pytest = "^7.3.1"
-hypothesis = "^6.72.0"
+hypothesis = "^6.74.0"
+pytest-xdist = "^3.2.1"
+
+[tool.poetry.extras]
+pyaiken = ["pyaiken"]
 
 [build-system]
 requires = ["poetry-core"]

src/utils/mock.py

@@ -0,0 +1,189 @@
+from collections import defaultdict
+from typing import Any, Callable, Dict, List, Optional, Union
+
+from pycardano import (
+    Address,
+    ChainContext,
+    ExecutionUnits,
+    GenesisParameters,
+    Network,
+    PaymentSigningKey,
+    PaymentVerificationKey,
+    ProtocolParameters,
+    ScriptType,
+    Transaction,
+    TransactionId,
+    TransactionInput,
+    TransactionOutput,
+    UTxO,
+    Value,
+)
+
+from src.utils.protocol_params import (
+    DEFAULT_GENESIS_PARAMETERS,
+    DEFAULT_PROTOCOL_PARAMETERS,
+)
+from src.utils.tx_tools import evaluate_script, generate_script_contexts_resolved
+
+
+class MockChainContext(ChainContext):
+    def __init__(
+        self,
+        protocol_param: Optional[ProtocolParameters] = None,
+        genesis_param: Optional[GenesisParameters] = None,
+    ):
+        self._protocol_param = (
+            protocol_param if protocol_param else DEFAULT_PROTOCOL_PARAMETERS
+        )
+        self._genesis_param = (
+            genesis_param if genesis_param else DEFAULT_GENESIS_PARAMETERS
+        )
+        self._utxo_state: Dict[str, List[UTxO]] = defaultdict(list)
+        self._address_lookup: Dict[UTxO, str] = {}
+        self._utxo_from_txid: Dict[TransactionId, Dict[int, UTxO]] = defaultdict(dict)
+        self._network = Network.TESTNET
+        self._epoch = 0
+        self._last_block_slot = 0
+        self.opshin_scripts: Dict[ScriptType, Callable[[Any, Any, Any], Any]] = {}
+
+    @property
+    def protocol_param(self) -> ProtocolParameters:
+        return self._protocol_param
+
+    @property
+    def genesis_param(self) -> GenesisParameters:
+        return self._genesis_param
+
+    @property
+    def network(self) -> Network:
+        return self._network
+
+    @property
+    def epoch(self) -> int:
+        return self._epoch
+
+    @property
+    def last_block_slot(self) -> int:
+        return self._last_block_slot
+
+    def _utxos(self, address: str) -> List[UTxO]:
+        return self._utxo_state.get(address, [])
+
+    def add_utxo(self, utxo: UTxO):
+        address = str(utxo.output.address)
+        self._utxo_state[address].append(utxo)
+        self._address_lookup[utxo] = address
+        self._utxo_from_txid[utxo.input.transaction_id][utxo.input.index] = utxo
+
+    def get_address(self, utxo: UTxO) -> str:
+        return self._address_lookup[utxo]
+
+    def remove_utxo(self, utxo: UTxO):
+        del self._utxo_from_txid[utxo.input.transaction_id][utxo.input.index]
+        address = self._address_lookup[utxo]
+        del self._address_lookup[utxo]
+        i = self._utxo_state[address].index(utxo)
+        self._utxo_state[address].pop(i)
+
+    def get_utxo_from_txid(self, transaction_id: TransactionId, index: int) -> UTxO:
+        return self._utxo_from_txid[transaction_id][index]
+
+    def submit_tx(self, tx: Transaction):
+        self.evaluate_tx(tx)
+        self.submit_tx_mock(tx)
+
+    def submit_tx_mock(self, tx: Transaction):
+        for input in tx.transaction_body.inputs:
+            utxo = self.get_utxo_from_txid(input.transaction_id, input.index)
+            self.remove_utxo(utxo)
+        for i, output in enumerate(tx.transaction_body.outputs):
+            utxo = UTxO(TransactionInput(tx.id, i), output)
+            self.add_utxo(utxo)
+
+    def submit_tx_cbor(self, cbor: Union[bytes, str]):
+        return self.submit_tx(Transaction.from_cbor(cbor))
+
+    def evaluate_tx(self, tx: Transaction) -> Dict[str, ExecutionUnits]:
+        input_utxos = [
+            self.get_utxo_from_txid(input.transaction_id, input.index)
+            for input in tx.transaction_body.inputs
+        ]
+        ref_input_utxos = (
+            [
+                self.get_utxo_from_txid(input.transaction_id, input.index)
+                for input in tx.transaction_body.reference_inputs
+            ]
+            if tx.transaction_body.reference_inputs is not None
+            else []
+        )
+        script_invocations = generate_script_contexts_resolved(
+            tx, input_utxos, ref_input_utxos, lambda s: self.posix_from_slot(s)
+        )
+        ret = {}
+        for invocation in script_invocations:
+            # run opshin script if available
+            if self.opshin_scripts.get(invocation.script) is not None:
+                opshin_validator = self.opshin_scripts[invocation.script]
+                opshin_validator(
+                    invocation.datum,
+                    invocation.redeemer.data,
+                    invocation.script_context,
+                )
+            redeemer = invocation.redeemer
+            if redeemer.ex_units.steps <= 0 and redeemer.ex_units.mem <= 0:
+                redeemer.ex_units = ExecutionUnits(
+                    self.protocol_param.max_tx_ex_mem,
+                    self.protocol_param.max_tx_ex_steps,
+                )
+
+            (suc, err), (cpu, mem), logs = evaluate_script(invocation)
+            if err:
+                raise ValueError(err, logs)
+            key = f"{redeemer.tag.name.lower()}:{redeemer.index}"
+            ret[key] = ExecutionUnits(mem, cpu)
+        return ret
+
+    def evaluate_tx_cbor(self, cbor: Union[bytes, str]) -> Dict[str, ExecutionUnits]:
+        return self.evaluate_tx(Transaction.from_cbor(cbor))
+
+    def wait(self, slots):
+        self._last_block_slot += slots
+
+    def posix_from_slot(self, slot: int) -> int:
+        """Convert a slot to POSIX time (seconds)"""
+        return self.genesis_param.system_start + self.genesis_param.slot_length * slot
+
+    def slot_from_posix(self, posix: int) -> int:
+        """Convert POSIX time (seconds) to the last slot"""
+        return (
+            posix - self.genesis_param.system_start
+        ) // self.genesis_param.slot_length
+
+
+class MockUser:
+    def __init__(self, context: MockChainContext):
+        self.context = context
+        self.signing_key = PaymentSigningKey.generate()
+        self.verification_key = PaymentVerificationKey.from_signing_key(
+            self.signing_key
+        )
+        self.network = Network.TESTNET
+        self.address = Address(
+            payment_part=self.verification_key.hash(), network=self.network
+        )
+
+    def fund(self, amount: Union[int, Value]):
+        if isinstance(amount, int):
+            value = Value(coin=amount)
+        else:
+            value = amount
+        self.context.add_utxo(
+            # not sure what the correct genesis transaction is
+            UTxO(
+                TransactionInput(TransactionId(self.verification_key.payload), 0),
+                TransactionOutput(self.address, value),
+            ),
+        )
+
+    def utxos(self):
+        return self.context.utxos(self.address)