Skip to content

Ethereum Test Specs package

Test spec definitions and utilities.

BaseTest

Bases: BaseModel

Represents a base Ethereum test which must return a single test fixture.

Source code in src/ethereum_test_specs/base.py 
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
class BaseTest(BaseModel):
    """Represents a base Ethereum test which must return a single test fixture."""

    tag: str = ""

    _request: pytest.FixtureRequest | None = PrivateAttr(None)

    # Transition tool specific fields
    t8n_dump_dir: Path | None = Field(None, exclude=True)
    t8n_call_counter: int = Field(0, exclude=True)

    supported_fixture_formats: ClassVar[Sequence[FixtureFormat | LabeledFixtureFormat]] = []
    supported_execute_formats: ClassVar[Sequence[LabeledExecuteFormat]] = []

    supported_markers: ClassVar[Dict[str, str]] = {}

    @classmethod
    def discard_fixture_format_by_marks(
        cls,
        fixture_format: FixtureFormat,
        fork: Fork,
        markers: List[pytest.Mark],
    ) -> bool:
        """Discard a fixture format from filling if the appropriate marker is used."""
        return False

    @classmethod
    def from_test(
        cls: Type[T],
        *,
        base_test: "BaseTest",
        **kwargs,
    ) -> T:
        """Create a test in a different format from a base test."""
        new_instance = cls(
            tag=base_test.tag,
            t8n_dump_dir=base_test.t8n_dump_dir,
            **kwargs,
        )
        new_instance._request = base_test._request
        return new_instance

    @classmethod
    def discard_execute_format_by_marks(
        cls,
        execute_format: ExecuteFormat,
        fork: Fork,
        markers: List[pytest.Mark],
    ) -> bool:
        """Discard an execute format from executing if the appropriate marker is used."""
        return False

    @abstractmethod
    def generate(
        self,
        *,
        t8n: TransitionTool,
        fork: Fork,
        fixture_format: FixtureFormat,
        eips: Optional[List[int]] = None,
    ) -> BaseFixture:
        """Generate the list of test fixtures."""
        pass

    def execute(
        self,
        *,
        fork: Fork,
        execute_format: ExecuteFormat,
        eips: Optional[List[int]] = None,
    ) -> BaseExecute:
        """Generate the list of test fixtures."""
        raise Exception(f"Unsupported execute format: {execute_format}")

    @classmethod
    def pytest_parameter_name(cls) -> str:
        """
        Must return the name of the parameter used in pytest to select this
        spec type as filler for the test.

        By default, it returns the underscore separated name of the class.
        """
        return reduce(lambda x, y: x + ("_" if y.isupper() else "") + y, cls.__name__).lower()

    def get_next_transition_tool_output_path(self) -> str:
        """Return path to the next transition tool output file."""
        if not self.t8n_dump_dir:
            return ""
        current_value = self.t8n_call_counter
        self.t8n_call_counter += 1
        return path.join(
            self.t8n_dump_dir,
            str(current_value),
        )

    def is_slow_test(self) -> bool:
        """Check if the test is slow."""
        if self._request is not None and hasattr(self._request, "node"):
            return self._request.node.get_closest_marker("slow") is not None
        return False

    def is_exception_test(self) -> bool | None:
        """
        Check if the test is an exception test (invalid block, invalid transaction).

        `None` is returned if it's not possible to determine if the test is negative or not.
        This is the case when the test is not run in pytest.
        """
        if self._request is not None and hasattr(self._request, "node"):
            return self._request.node.get_closest_marker("exception_test") is not None
        return None

    def node_id(self) -> str:
        """Return the node ID of the test."""
        if self._request is not None and hasattr(self._request, "node"):
            return self._request.node.nodeid
        return ""

    def check_exception_test(
        self,
        *,
        exception: bool,
    ):
        """Compare the test marker against the outcome of the test."""
        negative_test_marker = self.is_exception_test()
        if negative_test_marker is None:
            return
        if negative_test_marker != exception:
            if exception:
                raise Exception(
                    "Test produced an invalid block or transaction but was not marked with the "
                    "`exception_test` marker. Add the `@pytest.mark.exception_test` decorator "
                    "to the test."
                )
            else:
                raise Exception(
                    "Test didn't produce an invalid block or transaction but was marked with the "
                    "`exception_test` marker. Remove the `@pytest.mark.exception_test` decorator "
                    "from the test."
                )

discard_fixture_format_by_marks(fixture_format, fork, markers) classmethod

Discard a fixture format from filling if the appropriate marker is used.

Source code in src/ethereum_test_specs/base.py 
63
64
65
66
67
68
69
70
71
@classmethod
def discard_fixture_format_by_marks(
    cls,
    fixture_format: FixtureFormat,
    fork: Fork,
    markers: List[pytest.Mark],
) -> bool:
    """Discard a fixture format from filling if the appropriate marker is used."""
    return False

from_test(*, base_test, **kwargs) classmethod

Create a test in a different format from a base test.

Source code in src/ethereum_test_specs/base.py 
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
@classmethod
def from_test(
    cls: Type[T],
    *,
    base_test: "BaseTest",
    **kwargs,
) -> T:
    """Create a test in a different format from a base test."""
    new_instance = cls(
        tag=base_test.tag,
        t8n_dump_dir=base_test.t8n_dump_dir,
        **kwargs,
    )
    new_instance._request = base_test._request
    return new_instance

discard_execute_format_by_marks(execute_format, fork, markers) classmethod

Discard an execute format from executing if the appropriate marker is used.

Source code in src/ethereum_test_specs/base.py 
89
90
91
92
93
94
95
96
97
@classmethod
def discard_execute_format_by_marks(
    cls,
    execute_format: ExecuteFormat,
    fork: Fork,
    markers: List[pytest.Mark],
) -> bool:
    """Discard an execute format from executing if the appropriate marker is used."""
    return False

generate(*, t8n, fork, fixture_format, eips=None) abstractmethod

Generate the list of test fixtures.

Source code in src/ethereum_test_specs/base.py 
 99
100
101
102
103
104
105
106
107
108
109
@abstractmethod
def generate(
    self,
    *,
    t8n: TransitionTool,
    fork: Fork,
    fixture_format: FixtureFormat,
    eips: Optional[List[int]] = None,
) -> BaseFixture:
    """Generate the list of test fixtures."""
    pass

execute(*, fork, execute_format, eips=None)

Generate the list of test fixtures.

Source code in src/ethereum_test_specs/base.py 
111
112
113
114
115
116
117
118
119
def execute(
    self,
    *,
    fork: Fork,
    execute_format: ExecuteFormat,
    eips: Optional[List[int]] = None,
) -> BaseExecute:
    """Generate the list of test fixtures."""
    raise Exception(f"Unsupported execute format: {execute_format}")

pytest_parameter_name() classmethod

Must return the name of the parameter used in pytest to select this spec type as filler for the test.

By default, it returns the underscore separated name of the class.

Source code in src/ethereum_test_specs/base.py 
121
122
123
124
125
126
127
128
129
@classmethod
def pytest_parameter_name(cls) -> str:
    """
    Must return the name of the parameter used in pytest to select this
    spec type as filler for the test.

    By default, it returns the underscore separated name of the class.
    """
    return reduce(lambda x, y: x + ("_" if y.isupper() else "") + y, cls.__name__).lower()

get_next_transition_tool_output_path()

Return path to the next transition tool output file.

Source code in src/ethereum_test_specs/base.py 
131
132
133
134
135
136
137
138
139
140
def get_next_transition_tool_output_path(self) -> str:
    """Return path to the next transition tool output file."""
    if not self.t8n_dump_dir:
        return ""
    current_value = self.t8n_call_counter
    self.t8n_call_counter += 1
    return path.join(
        self.t8n_dump_dir,
        str(current_value),
    )

is_slow_test()

Check if the test is slow.

Source code in src/ethereum_test_specs/base.py 
142
143
144
145
146
def is_slow_test(self) -> bool:
    """Check if the test is slow."""
    if self._request is not None and hasattr(self._request, "node"):
        return self._request.node.get_closest_marker("slow") is not None
    return False

is_exception_test()

Check if the test is an exception test (invalid block, invalid transaction).

None is returned if it's not possible to determine if the test is negative or not. This is the case when the test is not run in pytest.

Source code in src/ethereum_test_specs/base.py 
148
149
150
151
152
153
154
155
156
157
def is_exception_test(self) -> bool | None:
    """
    Check if the test is an exception test (invalid block, invalid transaction).

    `None` is returned if it's not possible to determine if the test is negative or not.
    This is the case when the test is not run in pytest.
    """
    if self._request is not None and hasattr(self._request, "node"):
        return self._request.node.get_closest_marker("exception_test") is not None
    return None

node_id()

Return the node ID of the test.

Source code in src/ethereum_test_specs/base.py 
159
160
161
162
163
def node_id(self) -> str:
    """Return the node ID of the test."""
    if self._request is not None and hasattr(self._request, "node"):
        return self._request.node.nodeid
    return ""

check_exception_test(*, exception)

Compare the test marker against the outcome of the test.

Source code in src/ethereum_test_specs/base.py 
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
def check_exception_test(
    self,
    *,
    exception: bool,
):
    """Compare the test marker against the outcome of the test."""
    negative_test_marker = self.is_exception_test()
    if negative_test_marker is None:
        return
    if negative_test_marker != exception:
        if exception:
            raise Exception(
                "Test produced an invalid block or transaction but was not marked with the "
                "`exception_test` marker. Add the `@pytest.mark.exception_test` decorator "
                "to the test."
            )
        else:
            raise Exception(
                "Test didn't produce an invalid block or transaction but was marked with the "
                "`exception_test` marker. Remove the `@pytest.mark.exception_test` decorator "
                "from the test."
            )

BaseStaticTest

Bases: BaseModel

Represents a base class that reads cases from static files.

Source code in src/ethereum_test_specs/base_static.py 
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
class BaseStaticTest(BaseModel):
    """Represents a base class that reads cases from static files."""

    formats: ClassVar[List[Type["BaseStaticTest"]]] = []
    formats_type_adapter: ClassVar[TypeAdapter]

    format_name: ClassVar[str] = ""

    @classmethod
    def __pydantic_init_subclass__(cls, **kwargs):
        """
        Register all subclasses of BaseStaticTest with a static test format name set
        as possible static test format.
        """
        if cls.format_name:
            # Register the new fixture format
            BaseStaticTest.formats.append(cls)
            if len(BaseStaticTest.formats) > 1:
                BaseStaticTest.formats_type_adapter = TypeAdapter(
                    Union[tuple(BaseStaticTest.formats)],
                )
            else:
                BaseStaticTest.formats_type_adapter = TypeAdapter(cls)

    @model_validator(mode="wrap")
    @classmethod
    def _parse_into_subclass(
        cls, v: Any, handler: ValidatorFunctionWrapHandler
    ) -> "BaseStaticTest":
        """Parse the static test into the correct subclass."""
        if cls is BaseStaticTest:
            return BaseStaticTest.formats_type_adapter.validate_python(v)
        return handler(v)

    @abstractmethod
    def fill_function(self) -> Callable:
        """
        Return the test function that can be used to fill the test.

        This method should be implemented by the subclasses.

        The function returned can be optionally decorated with the `@pytest.mark.parametrize`
        decorator to parametrize the test with the number of sub test cases.

        Example:
        ```
        @pytest.mark.parametrize("n", [1])
        @pytest.mark.parametrize("m", [1, 2])
        @pytest.mark.valid_from("Homestead")
        def test_state_filler(
            state_test: StateTestFiller,
            fork: Fork,
            pre: Alloc,
            n: int,
            m: int,
        ):
            \"\"\"Generate a test from a static state filler.\"\"\"
            assert n == 1
            assert m in [1, 2]
            env = Environment(**self.env.model_dump())
            sender = pre.fund_eoa()
            tx = Transaction(
                ty=0x0,
                nonce=0,
                to=Address(0x1000),
                gas_limit=500000,
                protected=False if fork in [Frontier, Homestead] else True,
                data="",
                sender=sender,
            )
            state_test(env=env, pre=pre, post={}, tx=tx)

        return test_state_filler
        ```

        To aid the generation of the test, the function can be defined and then the decorator be
        applied after defining the function:

        ```
        def test_state_filler(
            state_test: StateTestFiller,
            fork: Fork,
            pre: Alloc,
            n: int,
            m: int,
        ):
            ...
        test_state_filler = pytest.mark.parametrize("n", [1])(test_state_filler)
        test_state_filler = pytest.mark.parametrize("m", [1, 2])(test_state_filler)
        if self.valid_from:
            test_state_filler = pytest.mark.valid_from(self.valid_from)(test_state_filler)
        if self.valid_until:
            test_state_filler = pytest.mark.valid_until(self.valid_until)(test_state_filler)
        return test_state_filler
        ```

        The function can contain the following parameters on top of the spec type parameter
        (`state_test` in the example above):
        - `fork`: The fork for which the test is currently being filled.
        - `pre`: The pre-state of the test.

        """
        raise NotImplementedError

    @staticmethod
    def remove_comments(data: Dict) -> Dict:
        """Remove comments from a dictionary."""
        result = {}
        for k, v in data.items():
            if isinstance(k, str) and k.startswith("//"):
                continue
            if isinstance(v, dict):
                v = BaseStaticTest.remove_comments(v)
            elif isinstance(v, list):
                v = [BaseStaticTest.remove_comments(i) if isinstance(i, dict) else i for i in v]
            result[k] = v
        return result

    @model_validator(mode="before")
    @classmethod
    def remove_comments_from_model(cls, data: Any) -> Any:
        """Remove comments from the static file loaded, if any."""
        if isinstance(data, dict):
            return BaseStaticTest.remove_comments(data)
        return data

__pydantic_init_subclass__(**kwargs) classmethod

Register all subclasses of BaseStaticTest with a static test format name set as possible static test format.

Source code in src/ethereum_test_specs/base_static.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
@classmethod
def __pydantic_init_subclass__(cls, **kwargs):
    """
    Register all subclasses of BaseStaticTest with a static test format name set
    as possible static test format.
    """
    if cls.format_name:
        # Register the new fixture format
        BaseStaticTest.formats.append(cls)
        if len(BaseStaticTest.formats) > 1:
            BaseStaticTest.formats_type_adapter = TypeAdapter(
                Union[tuple(BaseStaticTest.formats)],
            )
        else:
            BaseStaticTest.formats_type_adapter = TypeAdapter(cls)

fill_function() abstractmethod

Return the test function that can be used to fill the test.

This method should be implemented by the subclasses.

The function returned can be optionally decorated with the @pytest.mark.parametrize decorator to parametrize the test with the number of sub test cases.

Example: 

@pytest.mark.parametrize("n", [1])
@pytest.mark.parametrize("m", [1, 2])
@pytest.mark.valid_from("Homestead")
def test_state_filler(
    state_test: StateTestFiller,
    fork: Fork,
    pre: Alloc,
    n: int,
    m: int,
):
    """Generate a test from a static state filler."""
    assert n == 1
    assert m in [1, 2]
    env = Environment(**self.env.model_dump())
    sender = pre.fund_eoa()
    tx = Transaction(
        ty=0x0,
        nonce=0,
        to=Address(0x1000),
        gas_limit=500000,
        protected=False if fork in [Frontier, Homestead] else True,
        data="",
        sender=sender,
    )
    state_test(env=env, pre=pre, post={}, tx=tx)

return test_state_filler

To aid the generation of the test, the function can be defined and then the decorator be applied after defining the function:

def test_state_filler(
    state_test: StateTestFiller,
    fork: Fork,
    pre: Alloc,
    n: int,
    m: int,
):
    ...
test_state_filler = pytest.mark.parametrize("n", [1])(test_state_filler)
test_state_filler = pytest.mark.parametrize("m", [1, 2])(test_state_filler)
if self.valid_from:
    test_state_filler = pytest.mark.valid_from(self.valid_from)(test_state_filler)
if self.valid_until:
    test_state_filler = pytest.mark.valid_until(self.valid_until)(test_state_filler)
return test_state_filler

The function can contain the following parameters on top of the spec type parameter (state_test in the example above): - fork: The fork for which the test is currently being filled. - pre: The pre-state of the test.

Source code in src/ethereum_test_specs/base_static.py 
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
@abstractmethod
def fill_function(self) -> Callable:
    """
    Return the test function that can be used to fill the test.

    This method should be implemented by the subclasses.

    The function returned can be optionally decorated with the `@pytest.mark.parametrize`
    decorator to parametrize the test with the number of sub test cases.

    Example:
    ```
    @pytest.mark.parametrize("n", [1])
    @pytest.mark.parametrize("m", [1, 2])
    @pytest.mark.valid_from("Homestead")
    def test_state_filler(
        state_test: StateTestFiller,
        fork: Fork,
        pre: Alloc,
        n: int,
        m: int,
    ):
        \"\"\"Generate a test from a static state filler.\"\"\"
        assert n == 1
        assert m in [1, 2]
        env = Environment(**self.env.model_dump())
        sender = pre.fund_eoa()
        tx = Transaction(
            ty=0x0,
            nonce=0,
            to=Address(0x1000),
            gas_limit=500000,
            protected=False if fork in [Frontier, Homestead] else True,
            data="",
            sender=sender,
        )
        state_test(env=env, pre=pre, post={}, tx=tx)

    return test_state_filler
    ```

    To aid the generation of the test, the function can be defined and then the decorator be
    applied after defining the function:

    ```
    def test_state_filler(
        state_test: StateTestFiller,
        fork: Fork,
        pre: Alloc,
        n: int,
        m: int,
    ):
        ...
    test_state_filler = pytest.mark.parametrize("n", [1])(test_state_filler)
    test_state_filler = pytest.mark.parametrize("m", [1, 2])(test_state_filler)
    if self.valid_from:
        test_state_filler = pytest.mark.valid_from(self.valid_from)(test_state_filler)
    if self.valid_until:
        test_state_filler = pytest.mark.valid_until(self.valid_until)(test_state_filler)
    return test_state_filler
    ```

    The function can contain the following parameters on top of the spec type parameter
    (`state_test` in the example above):
    - `fork`: The fork for which the test is currently being filled.
    - `pre`: The pre-state of the test.

    """
    raise NotImplementedError

remove_comments(data) staticmethod

Remove comments from a dictionary.

Source code in src/ethereum_test_specs/base_static.py 
121
122
123
124
125
126
127
128
129
130
131
132
133
@staticmethod
def remove_comments(data: Dict) -> Dict:
    """Remove comments from a dictionary."""
    result = {}
    for k, v in data.items():
        if isinstance(k, str) and k.startswith("//"):
            continue
        if isinstance(v, dict):
            v = BaseStaticTest.remove_comments(v)
        elif isinstance(v, list):
            v = [BaseStaticTest.remove_comments(i) if isinstance(i, dict) else i for i in v]
        result[k] = v
    return result

remove_comments_from_model(data) classmethod

Remove comments from the static file loaded, if any.

Source code in src/ethereum_test_specs/base_static.py 
135
136
137
138
139
140
141
@model_validator(mode="before")
@classmethod
def remove_comments_from_model(cls, data: Any) -> Any:
    """Remove comments from the static file loaded, if any."""
    if isinstance(data, dict):
        return BaseStaticTest.remove_comments(data)
    return data

BlockchainTest

Bases: BaseTest

Filler type that tests multiple blocks (valid or invalid) in a chain.

Source code in src/ethereum_test_specs/blockchain.py 
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
class BlockchainTest(BaseTest):
    """Filler type that tests multiple blocks (valid or invalid) in a chain."""

    pre: Alloc
    post: Alloc
    blocks: List[Block]
    genesis_environment: Environment = Field(default_factory=Environment)
    verify_sync: bool = False
    chain_id: int = 1

    supported_fixture_formats: ClassVar[Sequence[FixtureFormat | LabeledFixtureFormat]] = [
        BlockchainFixture,
        BlockchainEngineFixture,
    ]
    supported_execute_formats: ClassVar[Sequence[LabeledExecuteFormat]] = [
        LabeledExecuteFormat(
            TransactionPost,
            "blockchain_test",
            "An execute test derived from a blockchain test",
        ),
    ]

    supported_markers: ClassVar[Dict[str, str]] = {
        "blockchain_test_engine_only": "Only generate a blockchain test engine fixture",
        "blockchain_test_only": "Only generate a blockchain test fixture",
    }

    @classmethod
    def discard_fixture_format_by_marks(
        cls,
        fixture_format: FixtureFormat,
        fork: Fork,
        markers: List[pytest.Mark],
    ) -> bool:
        """Discard a fixture format from filling if the appropriate marker is used."""
        if "blockchain_test_only" in [m.name for m in markers]:
            return fixture_format != BlockchainFixture
        if "blockchain_test_engine_only" in [m.name for m in markers]:
            return fixture_format != BlockchainEngineFixture
        return False

    @staticmethod
    def make_genesis(
        genesis_environment: Environment,
        pre: Alloc,
        fork: Fork,
    ) -> Tuple[Alloc, FixtureBlock]:
        """Create a genesis block from the blockchain test definition."""
        env = genesis_environment.set_fork_requirements(fork)
        assert env.withdrawals is None or len(env.withdrawals) == 0, (
            "withdrawals must be empty at genesis"
        )
        assert env.parent_beacon_block_root is None or env.parent_beacon_block_root == Hash(0), (
            "parent_beacon_block_root must be empty at genesis"
        )

        pre_alloc = Alloc.merge(
            Alloc.model_validate(fork.pre_allocation_blockchain()),
            pre,
        )
        if empty_accounts := pre_alloc.empty_accounts():
            raise Exception(f"Empty accounts in pre state: {empty_accounts}")
        state_root = pre_alloc.state_root()
        genesis = FixtureHeader(
            parent_hash=0,
            ommers_hash=EmptyOmmersRoot,
            fee_recipient=0,
            state_root=state_root,
            transactions_trie=EmptyTrieRoot,
            receipts_root=EmptyTrieRoot,
            logs_bloom=0,
            difficulty=0x20000 if env.difficulty is None else env.difficulty,
            number=0,
            gas_limit=env.gas_limit,
            gas_used=0,
            timestamp=0,
            extra_data=b"\x00",
            prev_randao=0,
            nonce=0,
            base_fee_per_gas=env.base_fee_per_gas,
            blob_gas_used=env.blob_gas_used,
            excess_blob_gas=env.excess_blob_gas,
            withdrawals_root=(
                Withdrawal.list_root(env.withdrawals) if env.withdrawals is not None else None
            ),
            parent_beacon_block_root=env.parent_beacon_block_root,
            requests_hash=Requests() if fork.header_requests_required(0, 0) else None,
            fork=fork,
        )

        return (
            pre_alloc,
            FixtureBlockBase(
                header=genesis,
                withdrawals=None if env.withdrawals is None else [],
            ).with_rlp(txs=[]),
        )

    def generate_block_data(
        self,
        t8n: TransitionTool,
        fork: Fork,
        block: Block,
        previous_env: Environment,
        previous_alloc: Alloc,
        eips: Optional[List[int]] = None,
        slow: bool = False,
    ) -> Tuple[FixtureHeader, List[Transaction], List[Bytes] | None, Alloc, Environment]:
        """Generate common block data for both make_fixture and make_hive_fixture."""
        if block.rlp and block.exception is not None:
            raise Exception(
                "test correctness: post-state cannot be verified if the "
                + "block's rlp is supplied and the block is not supposed "
                + "to produce an exception"
            )

        env = block.set_environment(previous_env)
        env = env.set_fork_requirements(fork)

        txs = [tx.with_signature_and_sender() for tx in block.txs]

        if failing_tx_count := len([tx for tx in txs if tx.error]) > 0:
            if failing_tx_count > 1:
                raise Exception(
                    "test correctness: only one transaction can produce an exception in a block"
                )
            if not txs[-1].error:
                raise Exception(
                    "test correctness: the transaction that produces an exception "
                    + "must be the last transaction in the block"
                )

        transition_tool_output = t8n.evaluate(
            alloc=previous_alloc,
            txs=txs,
            env=env,
            fork=fork,
            chain_id=self.chain_id,
            reward=fork.get_reward(env.number, env.timestamp),
            blob_schedule=fork.blob_schedule(),
            eips=eips,
            debug_output_path=self.get_next_transition_tool_output_path(),
            slow_request=slow,
        )

        try:
            rejected_txs = verify_transactions(
                txs=txs,
                result=transition_tool_output.result,
                transition_tool_exceptions_reliable=t8n.exception_mapper.reliable,
            )
            if (
                not rejected_txs
                and block.rlp_modifier is None
                and block.requests is None
                and not block.skip_exception_verification
            ):
                # Only verify block level exception if:
                # - No transaction exception was raised, because these are not reported as block
                #   exceptions.
                # - No RLP modifier was specified, because the modifier is what normally
                #   produces the block exception.
                # - No requests were specified, because modified requests are also what normally
                #   produces the block exception.
                verify_block(
                    block_number=env.number,
                    want_exception=block.exception,
                    result=transition_tool_output.result,
                    transition_tool_exceptions_reliable=t8n.exception_mapper.reliable,
                )
            verify_result(transition_tool_output.result, env)
        except Exception as e:
            print_traces(t8n.get_traces())
            pprint(transition_tool_output.result)
            pprint(previous_alloc)
            pprint(transition_tool_output.alloc)
            raise e

        if len(rejected_txs) > 0 and block.exception is None:
            print_traces(t8n.get_traces())
            raise Exception(
                "one or more transactions in `BlockchainTest` are "
                + "intrinsically invalid, but the block was not expected "
                + "to be invalid. Please verify whether the transaction "
                + "was indeed expected to fail and add the proper "
                + "`block.exception`"
            )

        # One special case of the invalid transactions is the blob gas used, since this value
        # is not included in the transition tool result, but it is included in the block header,
        # and some clients check it before executing the block by simply counting the type-3 txs,
        # we need to set the correct value by default.
        blob_gas_used: int | None = None
        if (blob_gas_per_blob := fork.blob_gas_per_blob(env.number, env.timestamp)) > 0:
            blob_gas_used = blob_gas_per_blob * count_blobs(txs)

        header = FixtureHeader(
            **(
                transition_tool_output.result.model_dump(
                    exclude_none=True, exclude={"blob_gas_used", "transactions_trie"}
                )
                | env.model_dump(exclude_none=True, exclude={"blob_gas_used"})
            ),
            blob_gas_used=blob_gas_used,
            transactions_trie=Transaction.list_root(txs),
            extra_data=block.extra_data if block.extra_data is not None else b"",
            fork=fork,
        )

        if block.header_verify is not None:
            # Verify the header after transition tool processing.
            block.header_verify.verify(header)

        requests_list: List[Bytes] | None = None
        if fork.header_requests_required(header.number, header.timestamp):
            assert transition_tool_output.result.requests is not None, (
                "Requests are required for this block"
            )
            requests = Requests(requests_lists=list(transition_tool_output.result.requests))

            if Hash(requests) != header.requests_hash:
                raise Exception(
                    "Requests root in header does not match the requests root in the transition "
                    "tool output: "
                    f"{header.requests_hash} != {Hash(requests)}"
                )

            requests_list = requests.requests_list

        if block.requests is not None:
            header.requests_hash = Hash(Requests(requests_lists=list(block.requests)))
            requests_list = block.requests

        if block.rlp_modifier is not None:
            # Modify any parameter specified in the `rlp_modifier` after
            # transition tool processing.
            header = block.rlp_modifier.apply(header)
            header.fork = fork  # Deleted during `apply` because `exclude=True`

        return (
            header,
            txs,
            requests_list,
            transition_tool_output.alloc,
            env,
        )

    @staticmethod
    def network_info(fork: Fork, eips: Optional[List[int]] = None):
        """Return fixture network information for the fork & EIP/s."""
        return (
            "+".join([fork.blockchain_test_network_name()] + [str(eip) for eip in eips])
            if eips
            else fork.blockchain_test_network_name()
        )

    def verify_post_state(self, t8n, t8n_state: Alloc, expected_state: Alloc | None = None):
        """Verify post alloc after all block/s or payload/s are generated."""
        try:
            if expected_state:
                expected_state.verify_post_alloc(t8n_state)
            else:
                self.post.verify_post_alloc(t8n_state)
        except Exception as e:
            print_traces(t8n.get_traces())
            raise e

    def make_fixture(
        self,
        t8n: TransitionTool,
        fork: Fork,
        eips: Optional[List[int]] = None,
    ) -> BlockchainFixture:
        """Create a fixture from the blockchain test definition."""
        fixture_blocks: List[FixtureBlock | InvalidFixtureBlock] = []

        pre, genesis = BlockchainTest.make_genesis(self.genesis_environment, self.pre, fork)

        alloc = pre
        env = environment_from_parent_header(genesis.header)
        head = genesis.header.block_hash
        invalid_blocks = 0
        for block in self.blocks:
            if block.rlp is None:
                # This is the most common case, the RLP needs to be constructed
                # based on the transactions to be included in the block.
                # Set the environment according to the block to execute.
                header, txs, _, new_alloc, new_env = self.generate_block_data(
                    t8n=t8n,
                    fork=fork,
                    block=block,
                    previous_env=env,
                    previous_alloc=alloc,
                    eips=eips,
                    slow=self.is_slow_test(),
                )
                fixture_block = FixtureBlockBase(
                    header=header,
                    txs=[FixtureTransaction.from_transaction(tx) for tx in txs],
                    ommers=[],
                    withdrawals=(
                        [FixtureWithdrawal.from_withdrawal(w) for w in new_env.withdrawals]
                        if new_env.withdrawals is not None
                        else None
                    ),
                    fork=fork,
                ).with_rlp(txs=txs)
                if block.exception is None:
                    fixture_blocks.append(fixture_block)
                    # Update env, alloc and last block hash for the next block.
                    alloc = new_alloc
                    env = apply_new_parent(new_env, header)
                    head = header.block_hash
                else:
                    fixture_blocks.append(
                        InvalidFixtureBlock(
                            rlp=fixture_block.rlp,
                            expect_exception=block.exception,
                            rlp_decoded=(
                                None
                                if BlockException.RLP_STRUCTURES_ENCODING in block.exception
                                else fixture_block.without_rlp()
                            ),
                        ),
                    )
                    invalid_blocks += 1
            else:
                assert block.exception is not None, (
                    "test correctness: if the block's rlp is hard-coded, "
                    + "the block is expected to produce an exception"
                )
                fixture_blocks.append(
                    InvalidFixtureBlock(
                        rlp=block.rlp,
                        expect_exception=block.exception,
                    ),
                )
                invalid_blocks += 1

            if block.expected_post_state:
                self.verify_post_state(
                    t8n, t8n_state=alloc, expected_state=block.expected_post_state
                )
        self.check_exception_test(exception=invalid_blocks > 0)
        self.verify_post_state(t8n, t8n_state=alloc)
        network_info = BlockchainTest.network_info(fork, eips)
        return BlockchainFixture(
            fork=network_info,
            genesis=genesis.header,
            genesis_rlp=genesis.rlp,
            blocks=fixture_blocks,
            last_block_hash=head,
            pre=pre,
            post_state=alloc,
            config=FixtureConfig(
                fork=network_info,
                blob_schedule=FixtureBlobSchedule.from_blob_schedule(fork.blob_schedule()),
                chain_id=self.chain_id,
            ),
        )

    def make_hive_fixture(
        self,
        t8n: TransitionTool,
        fork: Fork,
        eips: Optional[List[int]] = None,
    ) -> BlockchainEngineFixture:
        """Create a hive fixture from the blocktest definition."""
        fixture_payloads: List[FixtureEngineNewPayload] = []

        pre, genesis = BlockchainTest.make_genesis(self.genesis_environment, self.pre, fork)
        alloc = pre
        env = environment_from_parent_header(genesis.header)
        head_hash = genesis.header.block_hash
        invalid_blocks = 0
        for block in self.blocks:
            header, txs, requests, new_alloc, new_env = self.generate_block_data(
                t8n=t8n,
                fork=fork,
                block=block,
                previous_env=env,
                previous_alloc=alloc,
                eips=eips,
                slow=self.is_slow_test(),
            )
            if block.rlp is None:
                fixture_payloads.append(
                    FixtureEngineNewPayload.from_fixture_header(
                        fork=fork,
                        header=header,
                        transactions=txs,
                        withdrawals=new_env.withdrawals,
                        requests=requests,
                        validation_error=block.exception,
                        error_code=block.engine_api_error_code,
                    )
                )
                if block.exception is None:
                    alloc = new_alloc
                    env = apply_new_parent(env, header)
                    head_hash = header.block_hash
                else:
                    invalid_blocks += 1

            if block.expected_post_state:
                self.verify_post_state(
                    t8n, t8n_state=alloc, expected_state=block.expected_post_state
                )
        self.check_exception_test(exception=invalid_blocks > 0)
        fcu_version = fork.engine_forkchoice_updated_version(header.number, header.timestamp)
        assert fcu_version is not None, (
            "A hive fixture was requested but no forkchoice update is defined."
            " The framework should never try to execute this test case."
        )

        self.verify_post_state(t8n, t8n_state=alloc)

        sync_payload: Optional[FixtureEngineNewPayload] = None
        if self.verify_sync:
            # Test is marked for syncing verification.
            assert genesis.header.block_hash != head_hash, (
                "Invalid payload tests negative test via sync is not supported yet."
            )

            # Most clients require the header to start the sync process, so we create an empty
            # block on top of the last block of the test to send it as new payload and trigger the
            # sync process.
            sync_header, _, requests, _, _ = self.generate_block_data(
                t8n=t8n,
                fork=fork,
                block=Block(),
                previous_env=env,
                previous_alloc=alloc,
                eips=eips,
            )
            sync_payload = FixtureEngineNewPayload.from_fixture_header(
                fork=fork,
                header=sync_header,
                transactions=[],
                withdrawals=[],
                requests=requests,
                validation_error=None,
                error_code=None,
            )

        network_info = BlockchainTest.network_info(fork, eips)
        return BlockchainEngineFixture(
            fork=network_info,
            genesis=genesis.header,
            payloads=fixture_payloads,
            fcu_version=fcu_version,
            pre=pre,
            post_state=alloc,
            sync_payload=sync_payload,
            last_block_hash=head_hash,
            config=FixtureConfig(
                fork=network_info,
                chain_id=self.chain_id,
                blob_schedule=FixtureBlobSchedule.from_blob_schedule(fork.blob_schedule()),
            ),
        )

    def generate(
        self,
        t8n: TransitionTool,
        fork: Fork,
        fixture_format: FixtureFormat,
        eips: Optional[List[int]] = None,
    ) -> BaseFixture:
        """Generate the BlockchainTest fixture."""
        t8n.reset_traces()
        if fixture_format == BlockchainEngineFixture:
            return self.make_hive_fixture(t8n, fork, eips)
        elif fixture_format == BlockchainFixture:
            return self.make_fixture(t8n, fork, eips)

        raise Exception(f"Unknown fixture format: {fixture_format}")

    def execute(
        self,
        *,
        fork: Fork,
        execute_format: ExecuteFormat,
        eips: Optional[List[int]] = None,
    ) -> BaseExecute:
        """Generate the list of test fixtures."""
        if execute_format == TransactionPost:
            blocks: List[List[Transaction]] = []
            for block in self.blocks:
                blocks += [block.txs]
            return TransactionPost(
                blocks=blocks,
                post=self.post,
            )
        raise Exception(f"Unsupported execute format: {execute_format}")

discard_fixture_format_by_marks(fixture_format, fork, markers) classmethod

Discard a fixture format from filling if the appropriate marker is used.

Source code in src/ethereum_test_specs/blockchain.py 
320
321
322
323
324
325
326
327
328
329
330
331
332
@classmethod
def discard_fixture_format_by_marks(
    cls,
    fixture_format: FixtureFormat,
    fork: Fork,
    markers: List[pytest.Mark],
) -> bool:
    """Discard a fixture format from filling if the appropriate marker is used."""
    if "blockchain_test_only" in [m.name for m in markers]:
        return fixture_format != BlockchainFixture
    if "blockchain_test_engine_only" in [m.name for m in markers]:
        return fixture_format != BlockchainEngineFixture
    return False

make_genesis(genesis_environment, pre, fork) staticmethod

Create a genesis block from the blockchain test definition.

Source code in src/ethereum_test_specs/blockchain.py 
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
@staticmethod
def make_genesis(
    genesis_environment: Environment,
    pre: Alloc,
    fork: Fork,
) -> Tuple[Alloc, FixtureBlock]:
    """Create a genesis block from the blockchain test definition."""
    env = genesis_environment.set_fork_requirements(fork)
    assert env.withdrawals is None or len(env.withdrawals) == 0, (
        "withdrawals must be empty at genesis"
    )
    assert env.parent_beacon_block_root is None or env.parent_beacon_block_root == Hash(0), (
        "parent_beacon_block_root must be empty at genesis"
    )

    pre_alloc = Alloc.merge(
        Alloc.model_validate(fork.pre_allocation_blockchain()),
        pre,
    )
    if empty_accounts := pre_alloc.empty_accounts():
        raise Exception(f"Empty accounts in pre state: {empty_accounts}")
    state_root = pre_alloc.state_root()
    genesis = FixtureHeader(
        parent_hash=0,
        ommers_hash=EmptyOmmersRoot,
        fee_recipient=0,
        state_root=state_root,
        transactions_trie=EmptyTrieRoot,
        receipts_root=EmptyTrieRoot,
        logs_bloom=0,
        difficulty=0x20000 if env.difficulty is None else env.difficulty,
        number=0,
        gas_limit=env.gas_limit,
        gas_used=0,
        timestamp=0,
        extra_data=b"\x00",
        prev_randao=0,
        nonce=0,
        base_fee_per_gas=env.base_fee_per_gas,
        blob_gas_used=env.blob_gas_used,
        excess_blob_gas=env.excess_blob_gas,
        withdrawals_root=(
            Withdrawal.list_root(env.withdrawals) if env.withdrawals is not None else None
        ),
        parent_beacon_block_root=env.parent_beacon_block_root,
        requests_hash=Requests() if fork.header_requests_required(0, 0) else None,
        fork=fork,
    )

    return (
        pre_alloc,
        FixtureBlockBase(
            header=genesis,
            withdrawals=None if env.withdrawals is None else [],
        ).with_rlp(txs=[]),
    )

generate_block_data(t8n, fork, block, previous_env, previous_alloc, eips=None, slow=False)

Generate common block data for both make_fixture and make_hive_fixture.

Source code in src/ethereum_test_specs/blockchain.py 
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
def generate_block_data(
    self,
    t8n: TransitionTool,
    fork: Fork,
    block: Block,
    previous_env: Environment,
    previous_alloc: Alloc,
    eips: Optional[List[int]] = None,
    slow: bool = False,
) -> Tuple[FixtureHeader, List[Transaction], List[Bytes] | None, Alloc, Environment]:
    """Generate common block data for both make_fixture and make_hive_fixture."""
    if block.rlp and block.exception is not None:
        raise Exception(
            "test correctness: post-state cannot be verified if the "
            + "block's rlp is supplied and the block is not supposed "
            + "to produce an exception"
        )

    env = block.set_environment(previous_env)
    env = env.set_fork_requirements(fork)

    txs = [tx.with_signature_and_sender() for tx in block.txs]

    if failing_tx_count := len([tx for tx in txs if tx.error]) > 0:
        if failing_tx_count > 1:
            raise Exception(
                "test correctness: only one transaction can produce an exception in a block"
            )
        if not txs[-1].error:
            raise Exception(
                "test correctness: the transaction that produces an exception "
                + "must be the last transaction in the block"
            )

    transition_tool_output = t8n.evaluate(
        alloc=previous_alloc,
        txs=txs,
        env=env,
        fork=fork,
        chain_id=self.chain_id,
        reward=fork.get_reward(env.number, env.timestamp),
        blob_schedule=fork.blob_schedule(),
        eips=eips,
        debug_output_path=self.get_next_transition_tool_output_path(),
        slow_request=slow,
    )

    try:
        rejected_txs = verify_transactions(
            txs=txs,
            result=transition_tool_output.result,
            transition_tool_exceptions_reliable=t8n.exception_mapper.reliable,
        )
        if (
            not rejected_txs
            and block.rlp_modifier is None
            and block.requests is None
            and not block.skip_exception_verification
        ):
            # Only verify block level exception if:
            # - No transaction exception was raised, because these are not reported as block
            #   exceptions.
            # - No RLP modifier was specified, because the modifier is what normally
            #   produces the block exception.
            # - No requests were specified, because modified requests are also what normally
            #   produces the block exception.
            verify_block(
                block_number=env.number,
                want_exception=block.exception,
                result=transition_tool_output.result,
                transition_tool_exceptions_reliable=t8n.exception_mapper.reliable,
            )
        verify_result(transition_tool_output.result, env)
    except Exception as e:
        print_traces(t8n.get_traces())
        pprint(transition_tool_output.result)
        pprint(previous_alloc)
        pprint(transition_tool_output.alloc)
        raise e

    if len(rejected_txs) > 0 and block.exception is None:
        print_traces(t8n.get_traces())
        raise Exception(
            "one or more transactions in `BlockchainTest` are "
            + "intrinsically invalid, but the block was not expected "
            + "to be invalid. Please verify whether the transaction "
            + "was indeed expected to fail and add the proper "
            + "`block.exception`"
        )

    # One special case of the invalid transactions is the blob gas used, since this value
    # is not included in the transition tool result, but it is included in the block header,
    # and some clients check it before executing the block by simply counting the type-3 txs,
    # we need to set the correct value by default.
    blob_gas_used: int | None = None
    if (blob_gas_per_blob := fork.blob_gas_per_blob(env.number, env.timestamp)) > 0:
        blob_gas_used = blob_gas_per_blob * count_blobs(txs)

    header = FixtureHeader(
        **(
            transition_tool_output.result.model_dump(
                exclude_none=True, exclude={"blob_gas_used", "transactions_trie"}
            )
            | env.model_dump(exclude_none=True, exclude={"blob_gas_used"})
        ),
        blob_gas_used=blob_gas_used,
        transactions_trie=Transaction.list_root(txs),
        extra_data=block.extra_data if block.extra_data is not None else b"",
        fork=fork,
    )

    if block.header_verify is not None:
        # Verify the header after transition tool processing.
        block.header_verify.verify(header)

    requests_list: List[Bytes] | None = None
    if fork.header_requests_required(header.number, header.timestamp):
        assert transition_tool_output.result.requests is not None, (
            "Requests are required for this block"
        )
        requests = Requests(requests_lists=list(transition_tool_output.result.requests))

        if Hash(requests) != header.requests_hash:
            raise Exception(
                "Requests root in header does not match the requests root in the transition "
                "tool output: "
                f"{header.requests_hash} != {Hash(requests)}"
            )

        requests_list = requests.requests_list

    if block.requests is not None:
        header.requests_hash = Hash(Requests(requests_lists=list(block.requests)))
        requests_list = block.requests

    if block.rlp_modifier is not None:
        # Modify any parameter specified in the `rlp_modifier` after
        # transition tool processing.
        header = block.rlp_modifier.apply(header)
        header.fork = fork  # Deleted during `apply` because `exclude=True`

    return (
        header,
        txs,
        requests_list,
        transition_tool_output.alloc,
        env,
    )

network_info(fork, eips=None) staticmethod

Return fixture network information for the fork & EIP/s.

Source code in src/ethereum_test_specs/blockchain.py 
540
541
542
543
544
545
546
547
@staticmethod
def network_info(fork: Fork, eips: Optional[List[int]] = None):
    """Return fixture network information for the fork & EIP/s."""
    return (
        "+".join([fork.blockchain_test_network_name()] + [str(eip) for eip in eips])
        if eips
        else fork.blockchain_test_network_name()
    )

verify_post_state(t8n, t8n_state, expected_state=None)

Verify post alloc after all block/s or payload/s are generated.

Source code in src/ethereum_test_specs/blockchain.py 
549
550
551
552
553
554
555
556
557
558
def verify_post_state(self, t8n, t8n_state: Alloc, expected_state: Alloc | None = None):
    """Verify post alloc after all block/s or payload/s are generated."""
    try:
        if expected_state:
            expected_state.verify_post_alloc(t8n_state)
        else:
            self.post.verify_post_alloc(t8n_state)
    except Exception as e:
        print_traces(t8n.get_traces())
        raise e

make_fixture(t8n, fork, eips=None)

Create a fixture from the blockchain test definition.

Source code in src/ethereum_test_specs/blockchain.py 
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
def make_fixture(
    self,
    t8n: TransitionTool,
    fork: Fork,
    eips: Optional[List[int]] = None,
) -> BlockchainFixture:
    """Create a fixture from the blockchain test definition."""
    fixture_blocks: List[FixtureBlock | InvalidFixtureBlock] = []

    pre, genesis = BlockchainTest.make_genesis(self.genesis_environment, self.pre, fork)

    alloc = pre
    env = environment_from_parent_header(genesis.header)
    head = genesis.header.block_hash
    invalid_blocks = 0
    for block in self.blocks:
        if block.rlp is None:
            # This is the most common case, the RLP needs to be constructed
            # based on the transactions to be included in the block.
            # Set the environment according to the block to execute.
            header, txs, _, new_alloc, new_env = self.generate_block_data(
                t8n=t8n,
                fork=fork,
                block=block,
                previous_env=env,
                previous_alloc=alloc,
                eips=eips,
                slow=self.is_slow_test(),
            )
            fixture_block = FixtureBlockBase(
                header=header,
                txs=[FixtureTransaction.from_transaction(tx) for tx in txs],
                ommers=[],
                withdrawals=(
                    [FixtureWithdrawal.from_withdrawal(w) for w in new_env.withdrawals]
                    if new_env.withdrawals is not None
                    else None
                ),
                fork=fork,
            ).with_rlp(txs=txs)
            if block.exception is None:
                fixture_blocks.append(fixture_block)
                # Update env, alloc and last block hash for the next block.
                alloc = new_alloc
                env = apply_new_parent(new_env, header)
                head = header.block_hash
            else:
                fixture_blocks.append(
                    InvalidFixtureBlock(
                        rlp=fixture_block.rlp,
                        expect_exception=block.exception,
                        rlp_decoded=(
                            None
                            if BlockException.RLP_STRUCTURES_ENCODING in block.exception
                            else fixture_block.without_rlp()
                        ),
                    ),
                )
                invalid_blocks += 1
        else:
            assert block.exception is not None, (
                "test correctness: if the block's rlp is hard-coded, "
                + "the block is expected to produce an exception"
            )
            fixture_blocks.append(
                InvalidFixtureBlock(
                    rlp=block.rlp,
                    expect_exception=block.exception,
                ),
            )
            invalid_blocks += 1

        if block.expected_post_state:
            self.verify_post_state(
                t8n, t8n_state=alloc, expected_state=block.expected_post_state
            )
    self.check_exception_test(exception=invalid_blocks > 0)
    self.verify_post_state(t8n, t8n_state=alloc)
    network_info = BlockchainTest.network_info(fork, eips)
    return BlockchainFixture(
        fork=network_info,
        genesis=genesis.header,
        genesis_rlp=genesis.rlp,
        blocks=fixture_blocks,
        last_block_hash=head,
        pre=pre,
        post_state=alloc,
        config=FixtureConfig(
            fork=network_info,
            blob_schedule=FixtureBlobSchedule.from_blob_schedule(fork.blob_schedule()),
            chain_id=self.chain_id,
        ),
    )

make_hive_fixture(t8n, fork, eips=None)

Create a hive fixture from the blocktest definition.

Source code in src/ethereum_test_specs/blockchain.py 
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
def make_hive_fixture(
    self,
    t8n: TransitionTool,
    fork: Fork,
    eips: Optional[List[int]] = None,
) -> BlockchainEngineFixture:
    """Create a hive fixture from the blocktest definition."""
    fixture_payloads: List[FixtureEngineNewPayload] = []

    pre, genesis = BlockchainTest.make_genesis(self.genesis_environment, self.pre, fork)
    alloc = pre
    env = environment_from_parent_header(genesis.header)
    head_hash = genesis.header.block_hash
    invalid_blocks = 0
    for block in self.blocks:
        header, txs, requests, new_alloc, new_env = self.generate_block_data(
            t8n=t8n,
            fork=fork,
            block=block,
            previous_env=env,
            previous_alloc=alloc,
            eips=eips,
            slow=self.is_slow_test(),
        )
        if block.rlp is None:
            fixture_payloads.append(
                FixtureEngineNewPayload.from_fixture_header(
                    fork=fork,
                    header=header,
                    transactions=txs,
                    withdrawals=new_env.withdrawals,
                    requests=requests,
                    validation_error=block.exception,
                    error_code=block.engine_api_error_code,
                )
            )
            if block.exception is None:
                alloc = new_alloc
                env = apply_new_parent(env, header)
                head_hash = header.block_hash
            else:
                invalid_blocks += 1

        if block.expected_post_state:
            self.verify_post_state(
                t8n, t8n_state=alloc, expected_state=block.expected_post_state
            )
    self.check_exception_test(exception=invalid_blocks > 0)
    fcu_version = fork.engine_forkchoice_updated_version(header.number, header.timestamp)
    assert fcu_version is not None, (
        "A hive fixture was requested but no forkchoice update is defined."
        " The framework should never try to execute this test case."
    )

    self.verify_post_state(t8n, t8n_state=alloc)

    sync_payload: Optional[FixtureEngineNewPayload] = None
    if self.verify_sync:
        # Test is marked for syncing verification.
        assert genesis.header.block_hash != head_hash, (
            "Invalid payload tests negative test via sync is not supported yet."
        )

        # Most clients require the header to start the sync process, so we create an empty
        # block on top of the last block of the test to send it as new payload and trigger the
        # sync process.
        sync_header, _, requests, _, _ = self.generate_block_data(
            t8n=t8n,
            fork=fork,
            block=Block(),
            previous_env=env,
            previous_alloc=alloc,
            eips=eips,
        )
        sync_payload = FixtureEngineNewPayload.from_fixture_header(
            fork=fork,
            header=sync_header,
            transactions=[],
            withdrawals=[],
            requests=requests,
            validation_error=None,
            error_code=None,
        )

    network_info = BlockchainTest.network_info(fork, eips)
    return BlockchainEngineFixture(
        fork=network_info,
        genesis=genesis.header,
        payloads=fixture_payloads,
        fcu_version=fcu_version,
        pre=pre,
        post_state=alloc,
        sync_payload=sync_payload,
        last_block_hash=head_hash,
        config=FixtureConfig(
            fork=network_info,
            chain_id=self.chain_id,
            blob_schedule=FixtureBlobSchedule.from_blob_schedule(fork.blob_schedule()),
        ),
    )

generate(t8n, fork, fixture_format, eips=None)

Generate the BlockchainTest fixture.

Source code in src/ethereum_test_specs/blockchain.py 
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
def generate(
    self,
    t8n: TransitionTool,
    fork: Fork,
    fixture_format: FixtureFormat,
    eips: Optional[List[int]] = None,
) -> BaseFixture:
    """Generate the BlockchainTest fixture."""
    t8n.reset_traces()
    if fixture_format == BlockchainEngineFixture:
        return self.make_hive_fixture(t8n, fork, eips)
    elif fixture_format == BlockchainFixture:
        return self.make_fixture(t8n, fork, eips)

    raise Exception(f"Unknown fixture format: {fixture_format}")

execute(*, fork, execute_format, eips=None)

Generate the list of test fixtures.

Source code in src/ethereum_test_specs/blockchain.py 
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
def execute(
    self,
    *,
    fork: Fork,
    execute_format: ExecuteFormat,
    eips: Optional[List[int]] = None,
) -> BaseExecute:
    """Generate the list of test fixtures."""
    if execute_format == TransactionPost:
        blocks: List[List[Transaction]] = []
        for block in self.blocks:
            blocks += [block.txs]
        return TransactionPost(
            blocks=blocks,
            post=self.post,
        )
    raise Exception(f"Unsupported execute format: {execute_format}")

EOFStateTest

Bases: EOFTest, Transaction

Filler type that generates an EOF test for container validation, and also tests the container during runtime using a state test (and blockchain test).

In the state or blockchain test, the container is first deployed to the pre-allocation and then a transaction is sent to the deployed container.

Container deployment/validation is not tested like in the EOFTest unless the container under test is an initcode container.

All fields from ethereum_test_types.Transaction are available for use in the test.

Source code in src/ethereum_test_specs/eof.py 
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
class EOFStateTest(EOFTest, Transaction):
    """
    Filler type that generates an EOF test for container validation, and also tests the container
    during runtime using a state test (and blockchain test).

    In the state or blockchain test, the container is first deployed to the pre-allocation and
    then a transaction is sent to the deployed container.

    Container deployment/validation is **not** tested like in the `EOFTest` unless the container
    under test is an initcode container.

    All fields from `ethereum_test_types.Transaction` are available for use in the test.
    """

    gas_limit: HexNumber = Field(HexNumber(10_000_000), serialization_alias="gas")
    """
    Gas limit for the transaction that deploys the container.
    """
    tx_sender_funding_amount: int = 1_000_000_000_000_000_000_000
    """
    Amount of funds to send to the sender EOA before the transaction.
    """
    env: Environment = Field(default_factory=Environment)
    """
    Environment object that is used during State Test generation.
    """
    container_post: Account = Field(default_factory=Account)
    """
    Account object used to verify the container post state.
    """

    supported_fixture_formats: ClassVar[Sequence[FixtureFormat | LabeledFixtureFormat]] = [
        EOFFixture
    ] + [
        LabeledFixtureFormat(
            fixture_format,
            f"eof_{fixture_format.format_name}",
            f"Tests that generate an EOF {fixture_format.format_name}.",
        )
        for fixture_format in StateTest.supported_fixture_formats
    ]

    supported_execute_formats: ClassVar[Sequence[LabeledExecuteFormat]] = [
        LabeledExecuteFormat(
            execute_format,
            f"eof_{execute_format.label}",
            f"Tests that generate an EOF {execute_format.label}.",
        )
        for execute_format in StateTest.supported_execute_formats
    ]

    @classmethod
    def pytest_parameter_name(cls) -> str:
        """Workaround for pytest parameter name."""
        return "eof_state_test"

    def model_post_init(self, __context):
        """Prepare the transaction parameters required to fill the test."""
        assert self.pre is not None, "pre must be set to generate a StateTest."

        EOFTest.model_post_init(self, __context)

        self.sender = self.pre.fund_eoa(amount=self.tx_sender_funding_amount)
        if self.post is None:
            self.post = Alloc()

        if self.expect_exception is not None and self.container_kind == ContainerKind.RUNTIME:
            # Invalid EOF runtime code
            initcode = Container.Init(deploy_container=self.container)
            self.to = self.pre.deploy_contract(
                Op.TXCREATE(tx_initcode_hash=initcode.hash) + Op.STOP
            )
            self.initcodes = [initcode]

            # Run transaction model validation
            Transaction.model_post_init(self, __context)

            self.post[compute_eofcreate_address(self.to, 0)] = None  # Expect failure.
        elif self.expect_exception is not None and self.container_kind == ContainerKind.INITCODE:
            # Invalid EOF initcode
            self.to = self.pre.deploy_contract(
                Op.TXCREATE(tx_initcode_hash=self.container.hash) + Op.STOP
            )
            self.initcodes = [self.container]

            # Run transaction model validation
            Transaction.model_post_init(self, __context)

            self.post[compute_eofcreate_address(self.to, 0)] = None  # Expect failure.
        elif self.container_kind == ContainerKind.INITCODE:
            self.to = self.pre.deploy_contract(
                Op.TXCREATE(tx_initcode_hash=self.container.hash) + Op.STOP
            )
            self.initcodes = [self.container]

            # Run transaction model validation
            Transaction.model_post_init(self, __context)

            self.post[compute_eofcreate_address(self.to, 0)] = self.container_post
        else:
            self.to = self.pre.deploy_contract(code=self.container)

            # Run transaction model validation
            Transaction.model_post_init(self, __context)

            self.post[self.to] = self.container_post

    def generate_state_test(self, fork: Fork) -> StateTest:
        """Generate the StateTest filler."""
        assert self.pre is not None, "pre must be set to generate a StateTest."
        assert self.post is not None, "post must be set to generate a StateTest."

        return StateTest.from_test(
            base_test=self,
            pre=self.pre,
            tx=self,
            env=self.env,
            post=self.post,
        )

    def generate(
        self,
        *,
        t8n: TransitionTool,
        fork: Fork,
        eips: Optional[List[int]] = None,
        fixture_format: FixtureFormat,
        **_,
    ) -> BaseFixture:
        """Generate the BlockchainTest fixture."""
        if fixture_format == EOFFixture:
            if Bytes(self.container) in existing_tests:
                # Gracefully skip duplicate tests because one EOFStateTest can generate multiple
                # state fixtures with the same data.
                pytest.skip(f"Duplicate EOF container on EOFStateTest: {self.node_id()}")
            return self.make_eof_test_fixture(fork=fork, eips=eips)
        elif fixture_format in StateTest.supported_fixture_formats:
            return self.generate_state_test(fork).generate(
                t8n=t8n, fork=fork, fixture_format=fixture_format, eips=eips
            )

        raise Exception(f"Unknown fixture format: {fixture_format}")

gas_limit: HexNumber = Field(HexNumber(10000000), serialization_alias='gas') class-attribute instance-attribute

Gas limit for the transaction that deploys the container.

tx_sender_funding_amount: int = 1000000000000000000000 class-attribute instance-attribute

Amount of funds to send to the sender EOA before the transaction.

env: Environment = Field(default_factory=Environment) class-attribute instance-attribute

Environment object that is used during State Test generation.

container_post: Account = Field(default_factory=Account) class-attribute instance-attribute

Account object used to verify the container post state.

pytest_parameter_name() classmethod

Workaround for pytest parameter name.

Source code in src/ethereum_test_specs/eof.py 
562
563
564
565
@classmethod
def pytest_parameter_name(cls) -> str:
    """Workaround for pytest parameter name."""
    return "eof_state_test"

model_post_init(__context)

Prepare the transaction parameters required to fill the test.

Source code in src/ethereum_test_specs/eof.py 
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
def model_post_init(self, __context):
    """Prepare the transaction parameters required to fill the test."""
    assert self.pre is not None, "pre must be set to generate a StateTest."

    EOFTest.model_post_init(self, __context)

    self.sender = self.pre.fund_eoa(amount=self.tx_sender_funding_amount)
    if self.post is None:
        self.post = Alloc()

    if self.expect_exception is not None and self.container_kind == ContainerKind.RUNTIME:
        # Invalid EOF runtime code
        initcode = Container.Init(deploy_container=self.container)
        self.to = self.pre.deploy_contract(
            Op.TXCREATE(tx_initcode_hash=initcode.hash) + Op.STOP
        )
        self.initcodes = [initcode]

        # Run transaction model validation
        Transaction.model_post_init(self, __context)

        self.post[compute_eofcreate_address(self.to, 0)] = None  # Expect failure.
    elif self.expect_exception is not None and self.container_kind == ContainerKind.INITCODE:
        # Invalid EOF initcode
        self.to = self.pre.deploy_contract(
            Op.TXCREATE(tx_initcode_hash=self.container.hash) + Op.STOP
        )
        self.initcodes = [self.container]

        # Run transaction model validation
        Transaction.model_post_init(self, __context)

        self.post[compute_eofcreate_address(self.to, 0)] = None  # Expect failure.
    elif self.container_kind == ContainerKind.INITCODE:
        self.to = self.pre.deploy_contract(
            Op.TXCREATE(tx_initcode_hash=self.container.hash) + Op.STOP
        )
        self.initcodes = [self.container]

        # Run transaction model validation
        Transaction.model_post_init(self, __context)

        self.post[compute_eofcreate_address(self.to, 0)] = self.container_post
    else:
        self.to = self.pre.deploy_contract(code=self.container)

        # Run transaction model validation
        Transaction.model_post_init(self, __context)

        self.post[self.to] = self.container_post

generate_state_test(fork)

Generate the StateTest filler.

Source code in src/ethereum_test_specs/eof.py 
618
619
620
621
622
623
624
625
626
627
628
629
def generate_state_test(self, fork: Fork) -> StateTest:
    """Generate the StateTest filler."""
    assert self.pre is not None, "pre must be set to generate a StateTest."
    assert self.post is not None, "post must be set to generate a StateTest."

    return StateTest.from_test(
        base_test=self,
        pre=self.pre,
        tx=self,
        env=self.env,
        post=self.post,
    )

generate(*, t8n, fork, eips=None, fixture_format, **_)

Generate the BlockchainTest fixture.

Source code in src/ethereum_test_specs/eof.py 
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
def generate(
    self,
    *,
    t8n: TransitionTool,
    fork: Fork,
    eips: Optional[List[int]] = None,
    fixture_format: FixtureFormat,
    **_,
) -> BaseFixture:
    """Generate the BlockchainTest fixture."""
    if fixture_format == EOFFixture:
        if Bytes(self.container) in existing_tests:
            # Gracefully skip duplicate tests because one EOFStateTest can generate multiple
            # state fixtures with the same data.
            pytest.skip(f"Duplicate EOF container on EOFStateTest: {self.node_id()}")
        return self.make_eof_test_fixture(fork=fork, eips=eips)
    elif fixture_format in StateTest.supported_fixture_formats:
        return self.generate_state_test(fork).generate(
            t8n=t8n, fork=fork, fixture_format=fixture_format, eips=eips
        )

    raise Exception(f"Unknown fixture format: {fixture_format}")

EOFTest

Bases: BaseTest

Filler type that generates a test for EOF container validation.

A state test is also automatically generated where the container is wrapped in a contract-creating transaction to test deployment/validation on the instantiated blockchain.

Source code in src/ethereum_test_specs/eof.py 
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
class EOFTest(BaseTest):
    """
    Filler type that generates a test for EOF container validation.

    A state test is also automatically generated where the container is wrapped in a
    contract-creating transaction to test deployment/validation on the instantiated blockchain.
    """

    container: Container
    """
    EOF container that will be tested for validity.

    The only supported type at the moment is `ethereum_test_types.eof.v1.Container`.

    If an invalid container needs to be tested, and it cannot be generated using the
    Container class features, the `raw_bytes` field can be used to provide the raw
    container bytes.
    """
    expect_exception: EOFExceptionInstanceOrList | None = None
    """
    Expected exception that the container should raise when parsed by an EOF parser.

    Can be a single exception or a list of exceptions that the container is expected to raise,
    in which case the test will pass if any of the exceptions are raised.

    The list of supported exceptions can be found in the `ethereum_test_exceptions.EOFException`
    class.
    """
    container_kind: ContainerKind = ContainerKind.RUNTIME
    """
    Container kind type that the container should be treated as.

    The container kind can be one of the following:
    - `ContainerKind.INITCODE`: The container is an initcode container.
    - `ContainerKind.RUNTIME`: The container is a runtime container.

    The default value is `ContainerKind.RUNTIME`.
    """
    deployed_container: Container | None = None
    """
    To be used when the container is an initcode container and the expected deployed container is
    known.

    The value is only used when a State Test is generated from this EOF test to set the expected
    deployed container that should be found in the post state.

    If this field is not set, and the container is valid:
      - If the container kind is `ContainerKind.RUNTIME`, the deployed container is assumed to be
        the container itself, and an initcode container that wraps the container is generated
        automatically.
      - If the container kind is `ContainerKind.INITCODE`, `model_post_init` will attempt to infer
        the deployed container from the sections of the init-container, and the first
        container-type section will be used. An error will be raised if the deployed container
        cannot be inferred.

    If the value is set to `None`, it is assumed that the container is invalid and the test will
    expect that no contract is created.

    It is considered an error if:
      - The `deployed_container` field is set and the `container_kind` field is not set to
        `ContainerKind.INITCODE`.
      - The `deployed_container` field is set and the `expect_exception` is not `None`.

    The deployed container is **not** executed at any point during the EOF validation test nor
    the generated State Test. For container runtime testing use the `EOFStateTest` class.
    """
    pre: Alloc | None = None
    """
    Pre alloc object that is used during State Test generation.

    This field is automatically set by the test filler when generating a State Test from this EOF
    test and should otherwise be left unset.
    """
    post: Alloc | None = None
    """
    Post alloc object that is used during State Test generation.

    This field is automatically set by the test filler when generating a State Test from this EOF
    test and is normally not set by the user.
    """
    sender: EOA | None = None
    """
    Sender EOA object that is used during State Test generation.

    This field is automatically set by the `model_post_init` method and should otherwise be left
    unset.
    """

    supported_fixture_formats: ClassVar[Sequence[FixtureFormat | LabeledFixtureFormat]] = [
        EOFFixture
    ] + [
        LabeledFixtureFormat(
            fixture_format,
            f"{fixture_format.format_name}_from_eof_test",
            f"A {fixture_format.format_name} generated from an eof_test.",
        )
        for fixture_format in StateTest.supported_fixture_formats
    ]

    supported_execute_formats: ClassVar[Sequence[LabeledExecuteFormat]] = [
        LabeledExecuteFormat(
            execute_format,
            f"{execute_format.label}_from_eof_test",
            f"A {execute_format.label} generated from an eof_test.",
        )
        for execute_format in StateTest.supported_execute_formats
    ]

    supported_markers: ClassVar[Dict[str, str]] = {
        "eof_test_only": "Only generate an EOF test fixture",
    }

    @classmethod
    def discard_fixture_format_by_marks(
        cls,
        fixture_format: FixtureFormat,
        fork: Fork,
        markers: List[pytest.Mark],
    ) -> bool:
        """Discard a fixture format from filling if the appropriate marker is used."""
        if "eof_test_only" in [m.name for m in markers]:
            return fixture_format != EOFFixture
        return False

    @classmethod
    def pytest_parameter_name(cls) -> str:
        """Workaround for pytest parameter name."""
        return "eof_test"

    def model_post_init(self, __context):
        """Prepare the test exception based on the container."""
        if self.container.validity_error is not None:
            if self.expect_exception is not None:
                assert self.expect_exception == self.container.validity_error, (
                    f"Container validity error {self.container.validity_error} "
                    f"does not match expected exception {self.expect_exception}."
                )
            self.expect_exception = self.container.validity_error
            assert self.deployed_container is None, (
                "deployed_container must be None for invalid containers."
            )
        if "kind" in self.container.model_fields_set or "container_kind" in self.model_fields_set:
            if (
                "kind" in self.container.model_fields_set
                and "container_kind" in self.model_fields_set
            ):
                assert self.container.kind == self.container_kind, (
                    f"Container kind type {str(self.container.kind)} "
                    f"does not match test {self.container_kind}."
                )
            elif "kind" in self.container.model_fields_set:
                self.container_kind = self.container.kind
            elif "container_kind" in self.model_fields_set:
                self.container.kind = self.container_kind

        assert self.pre is not None, "pre must be set to generate a StateTest."
        self.sender = self.pre.fund_eoa()
        if self.post is None:
            self.post = Alloc()

    def make_eof_test_fixture(
        self,
        *,
        fork: Fork,
        eips: Optional[List[int]],
    ) -> EOFFixture:
        """Generate the EOF test fixture."""
        container_bytes = Bytes(self.container)
        if container_bytes in existing_tests:
            pytest.fail(
                f"Duplicate EOF test: {container_bytes}, "
                f"existing test: {existing_tests[container_bytes]}"
            )
        existing_tests[container_bytes] = self.node_id()
        vectors = [
            Vector(
                code=container_bytes,
                container_kind=self.container_kind,
                results={
                    fork.blockchain_test_network_name(): Result(
                        exception=self.expect_exception,
                        valid=self.expect_exception is None,
                    ),
                },
            )
        ]
        fixture = EOFFixture(vectors=dict(enumerate(vectors)))
        try:
            eof_parse = EOFParse()
        except FileNotFoundError as e:
            warnings.warn(
                f"{e} Skipping EOF fixture verification. Fixtures may be invalid!", stacklevel=2
            )
            return fixture

        for _, vector in fixture.vectors.items():
            expected_result = vector.results.get(fork.blockchain_test_network_name())
            if expected_result is None:
                raise Exception(f"EOF Fixture missing vector result for fork: {fork}")
            args = []
            if vector.container_kind == ContainerKind.INITCODE:
                args.append("--initcode")
            result = eof_parse.run(*args, input_value=str(vector.code))
            self.verify_result(result, expected_result, vector.code)

        return fixture

    def verify_result(self, result: CompletedProcess, expected_result: Result, code: Bytes):
        """Check that the reported exception string matches the expected error."""
        evmone_exception_mapper = EvmoneExceptionMapper()
        actual_exception_str = result.stdout.strip()
        actual_exception: EOFExceptionWithMessage | UndefinedException | None = None
        if not actual_exception_str.startswith("OK"):
            actual_exception = eof_exception_type_adapter.validate_python(
                actual_exception_str, context={"exception_mapper": evmone_exception_mapper}
            )

        if expected_result.exception is None:
            if actual_exception is not None:
                raise UnexpectedEOFExceptionError(code=code, got=f"{actual_exception}")
        else:
            expected_string = to_pipe_str(expected_result.exception)
            if actual_exception is None:
                raise ExpectedEOFExceptionError(
                    code=code,
                    expected=f"{expected_string}",
                )
            if (
                not isinstance(actual_exception, EOFExceptionWithMessage)
                or expected_result.exception not in actual_exception
            ):
                raise EOFExceptionMismatchError(
                    code=code,
                    expected=f"{expected_string}",
                    got=f"{actual_exception}",
                )

    def generate_eof_contract_create_transaction(self) -> Transaction:
        """Generate a transaction that creates a contract."""
        assert self.sender is not None, "sender must be set to generate a StateTest."
        assert self.post is not None, "post must be set to generate a StateTest."
        assert self.pre is not None, "pre must be set to generate a StateTest."

        initcode: Container
        deployed_container: Container | Bytes | None = None
        if self.container_kind == ContainerKind.INITCODE:
            initcode = self.container
            if "deployed_container" in self.model_fields_set:
                # In the case of an initcontainer where we know the deployed container,
                # we can use the initcontainer as-is.
                deployed_container = self.deployed_container
            elif self.expect_exception is None:
                # We have a valid init-container, but we don't know the deployed container.
                # Try to infer the deployed container from the sections of the init-container.
                assert self.container.raw_bytes is None, (
                    "deployed_container must be set for initcode containers with raw_bytes."
                )
                for section in self.container.sections:
                    if section.kind == SectionKind.CONTAINER:
                        deployed_container = section.data
                        break

                assert deployed_container is not None, (
                    "Unable to infer deployed container for init-container. "
                    "Use field `deployed_container` to set the expected deployed container."
                )
        else:
            assert self.deployed_container is None, (
                "deployed_container must be None for runtime containers."
            )
            initcode = Container(
                sections=[
                    Section.Code(Op.RETURNCODE[0](0, 0)),
                    Section.Container(self.container),
                ]
            )
            deployed_container = self.container

        factory_address = self.pre.deploy_contract(
            Op.TXCREATE(tx_initcode_hash=initcode.hash) + Op.STOP
        )

        tx = Transaction(
            sender=self.sender,
            to=factory_address,
            gas_limit=10_000_000,
            max_priority_fee_per_gas=10,
            max_fee_per_gas=10,
            initcodes=[initcode],
        )

        if self.expect_exception is not None or deployed_container is None:
            self.post[compute_eofcreate_address(factory_address, 0)] = None
        else:
            self.post[compute_eofcreate_address(factory_address, 0)] = Account(
                code=deployed_container,
            )
        return tx

    def generate_state_test(self, fork: Fork) -> StateTest:
        """Generate the StateTest filler."""
        return StateTest.from_test(
            base_test=self,
            pre=self.pre,
            tx=self.generate_eof_contract_create_transaction(),
            env=Environment(),
            post=self.post,
        )

    def generate(
        self,
        *,
        t8n: TransitionTool,
        fork: Fork,
        eips: Optional[List[int]] = None,
        fixture_format: FixtureFormat,
        **_,
    ) -> BaseFixture:
        """Generate the BlockchainTest fixture."""
        if fixture_format == EOFFixture:
            return self.make_eof_test_fixture(fork=fork, eips=eips)
        elif fixture_format in StateTest.supported_fixture_formats:
            return self.generate_state_test(fork).generate(
                t8n=t8n, fork=fork, fixture_format=fixture_format, eips=eips
            )
        raise Exception(f"Unknown fixture format: {fixture_format}")

    def execute(
        self,
        *,
        fork: Fork,
        execute_format: ExecuteFormat,
        eips: Optional[List[int]] = None,
    ) -> BaseExecute:
        """Generate the list of test fixtures."""
        if execute_format == TransactionPost:
            return self.generate_state_test(fork).execute(
                fork=fork, execute_format=execute_format, eips=eips
            )
        raise Exception(f"Unsupported execute format: {execute_format}")

container: Container instance-attribute

EOF container that will be tested for validity.

The only supported type at the moment is ethereum_test_types.eof.v1.Container.

If an invalid container needs to be tested, and it cannot be generated using the Container class features, the raw_bytes field can be used to provide the raw container bytes.

expect_exception: EOFExceptionInstanceOrList | None = None class-attribute instance-attribute

Expected exception that the container should raise when parsed by an EOF parser.

Can be a single exception or a list of exceptions that the container is expected to raise, in which case the test will pass if any of the exceptions are raised.

The list of supported exceptions can be found in the ethereum_test_exceptions.EOFException class.

container_kind: ContainerKind = ContainerKind.RUNTIME class-attribute instance-attribute

Container kind type that the container should be treated as.

The container kind can be one of the following: - ContainerKind.INITCODE: The container is an initcode container. - ContainerKind.RUNTIME: The container is a runtime container.

The default value is ContainerKind.RUNTIME.

deployed_container: Container | None = None class-attribute instance-attribute

To be used when the container is an initcode container and the expected deployed container is known.

The value is only used when a State Test is generated from this EOF test to set the expected deployed container that should be found in the post state.

If this field is not set, and the container is valid: - If the container kind is ContainerKind.RUNTIME, the deployed container is assumed to be the container itself, and an initcode container that wraps the container is generated automatically. - If the container kind is ContainerKind.INITCODE, model_post_init will attempt to infer the deployed container from the sections of the init-container, and the first container-type section will be used. An error will be raised if the deployed container cannot be inferred.

If the value is set to None, it is assumed that the container is invalid and the test will expect that no contract is created.

It is considered an error if
  • The deployed_container field is set and the container_kind field is not set to ContainerKind.INITCODE.
  • The deployed_container field is set and the expect_exception is not None.

The deployed container is not executed at any point during the EOF validation test nor the generated State Test. For container runtime testing use the EOFStateTest class.

pre: Alloc | None = None class-attribute instance-attribute

Pre alloc object that is used during State Test generation.

This field is automatically set by the test filler when generating a State Test from this EOF test and should otherwise be left unset.

post: Alloc | None = None class-attribute instance-attribute

Post alloc object that is used during State Test generation.

This field is automatically set by the test filler when generating a State Test from this EOF test and is normally not set by the user.

sender: EOA | None = None class-attribute instance-attribute

Sender EOA object that is used during State Test generation.

This field is automatically set by the model_post_init method and should otherwise be left unset.

discard_fixture_format_by_marks(fixture_format, fork, markers) classmethod

Discard a fixture format from filling if the appropriate marker is used.

Source code in src/ethereum_test_specs/eof.py 
277
278
279
280
281
282
283
284
285
286
287
@classmethod
def discard_fixture_format_by_marks(
    cls,
    fixture_format: FixtureFormat,
    fork: Fork,
    markers: List[pytest.Mark],
) -> bool:
    """Discard a fixture format from filling if the appropriate marker is used."""
    if "eof_test_only" in [m.name for m in markers]:
        return fixture_format != EOFFixture
    return False

pytest_parameter_name() classmethod

Workaround for pytest parameter name.

Source code in src/ethereum_test_specs/eof.py 
289
290
291
292
@classmethod
def pytest_parameter_name(cls) -> str:
    """Workaround for pytest parameter name."""
    return "eof_test"

model_post_init(__context)

Prepare the test exception based on the container.

Source code in src/ethereum_test_specs/eof.py 
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
def model_post_init(self, __context):
    """Prepare the test exception based on the container."""
    if self.container.validity_error is not None:
        if self.expect_exception is not None:
            assert self.expect_exception == self.container.validity_error, (
                f"Container validity error {self.container.validity_error} "
                f"does not match expected exception {self.expect_exception}."
            )
        self.expect_exception = self.container.validity_error
        assert self.deployed_container is None, (
            "deployed_container must be None for invalid containers."
        )
    if "kind" in self.container.model_fields_set or "container_kind" in self.model_fields_set:
        if (
            "kind" in self.container.model_fields_set
            and "container_kind" in self.model_fields_set
        ):
            assert self.container.kind == self.container_kind, (
                f"Container kind type {str(self.container.kind)} "
                f"does not match test {self.container_kind}."
            )
        elif "kind" in self.container.model_fields_set:
            self.container_kind = self.container.kind
        elif "container_kind" in self.model_fields_set:
            self.container.kind = self.container_kind

    assert self.pre is not None, "pre must be set to generate a StateTest."
    self.sender = self.pre.fund_eoa()
    if self.post is None:
        self.post = Alloc()

make_eof_test_fixture(*, fork, eips)

Generate the EOF test fixture.

Source code in src/ethereum_test_specs/eof.py 
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
def make_eof_test_fixture(
    self,
    *,
    fork: Fork,
    eips: Optional[List[int]],
) -> EOFFixture:
    """Generate the EOF test fixture."""
    container_bytes = Bytes(self.container)
    if container_bytes in existing_tests:
        pytest.fail(
            f"Duplicate EOF test: {container_bytes}, "
            f"existing test: {existing_tests[container_bytes]}"
        )
    existing_tests[container_bytes] = self.node_id()
    vectors = [
        Vector(
            code=container_bytes,
            container_kind=self.container_kind,
            results={
                fork.blockchain_test_network_name(): Result(
                    exception=self.expect_exception,
                    valid=self.expect_exception is None,
                ),
            },
        )
    ]
    fixture = EOFFixture(vectors=dict(enumerate(vectors)))
    try:
        eof_parse = EOFParse()
    except FileNotFoundError as e:
        warnings.warn(
            f"{e} Skipping EOF fixture verification. Fixtures may be invalid!", stacklevel=2
        )
        return fixture

    for _, vector in fixture.vectors.items():
        expected_result = vector.results.get(fork.blockchain_test_network_name())
        if expected_result is None:
            raise Exception(f"EOF Fixture missing vector result for fork: {fork}")
        args = []
        if vector.container_kind == ContainerKind.INITCODE:
            args.append("--initcode")
        result = eof_parse.run(*args, input_value=str(vector.code))
        self.verify_result(result, expected_result, vector.code)

    return fixture

verify_result(result, expected_result, code)

Check that the reported exception string matches the expected error.

Source code in src/ethereum_test_specs/eof.py 
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
def verify_result(self, result: CompletedProcess, expected_result: Result, code: Bytes):
    """Check that the reported exception string matches the expected error."""
    evmone_exception_mapper = EvmoneExceptionMapper()
    actual_exception_str = result.stdout.strip()
    actual_exception: EOFExceptionWithMessage | UndefinedException | None = None
    if not actual_exception_str.startswith("OK"):
        actual_exception = eof_exception_type_adapter.validate_python(
            actual_exception_str, context={"exception_mapper": evmone_exception_mapper}
        )

    if expected_result.exception is None:
        if actual_exception is not None:
            raise UnexpectedEOFExceptionError(code=code, got=f"{actual_exception}")
    else:
        expected_string = to_pipe_str(expected_result.exception)
        if actual_exception is None:
            raise ExpectedEOFExceptionError(
                code=code,
                expected=f"{expected_string}",
            )
        if (
            not isinstance(actual_exception, EOFExceptionWithMessage)
            or expected_result.exception not in actual_exception
        ):
            raise EOFExceptionMismatchError(
                code=code,
                expected=f"{expected_string}",
                got=f"{actual_exception}",
            )

generate_eof_contract_create_transaction()

Generate a transaction that creates a contract.

Source code in src/ethereum_test_specs/eof.py 
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
def generate_eof_contract_create_transaction(self) -> Transaction:
    """Generate a transaction that creates a contract."""
    assert self.sender is not None, "sender must be set to generate a StateTest."
    assert self.post is not None, "post must be set to generate a StateTest."
    assert self.pre is not None, "pre must be set to generate a StateTest."

    initcode: Container
    deployed_container: Container | Bytes | None = None
    if self.container_kind == ContainerKind.INITCODE:
        initcode = self.container
        if "deployed_container" in self.model_fields_set:
            # In the case of an initcontainer where we know the deployed container,
            # we can use the initcontainer as-is.
            deployed_container = self.deployed_container
        elif self.expect_exception is None:
            # We have a valid init-container, but we don't know the deployed container.
            # Try to infer the deployed container from the sections of the init-container.
            assert self.container.raw_bytes is None, (
                "deployed_container must be set for initcode containers with raw_bytes."
            )
            for section in self.container.sections:
                if section.kind == SectionKind.CONTAINER:
                    deployed_container = section.data
                    break

            assert deployed_container is not None, (
                "Unable to infer deployed container for init-container. "
                "Use field `deployed_container` to set the expected deployed container."
            )
    else:
        assert self.deployed_container is None, (
            "deployed_container must be None for runtime containers."
        )
        initcode = Container(
            sections=[
                Section.Code(Op.RETURNCODE[0](0, 0)),
                Section.Container(self.container),
            ]
        )
        deployed_container = self.container

    factory_address = self.pre.deploy_contract(
        Op.TXCREATE(tx_initcode_hash=initcode.hash) + Op.STOP
    )

    tx = Transaction(
        sender=self.sender,
        to=factory_address,
        gas_limit=10_000_000,
        max_priority_fee_per_gas=10,
        max_fee_per_gas=10,
        initcodes=[initcode],
    )

    if self.expect_exception is not None or deployed_container is None:
        self.post[compute_eofcreate_address(factory_address, 0)] = None
    else:
        self.post[compute_eofcreate_address(factory_address, 0)] = Account(
            code=deployed_container,
        )
    return tx

generate_state_test(fork)

Generate the StateTest filler.

Source code in src/ethereum_test_specs/eof.py 
464
465
466
467
468
469
470
471
472
def generate_state_test(self, fork: Fork) -> StateTest:
    """Generate the StateTest filler."""
    return StateTest.from_test(
        base_test=self,
        pre=self.pre,
        tx=self.generate_eof_contract_create_transaction(),
        env=Environment(),
        post=self.post,
    )

generate(*, t8n, fork, eips=None, fixture_format, **_)

Generate the BlockchainTest fixture.

Source code in src/ethereum_test_specs/eof.py 
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
def generate(
    self,
    *,
    t8n: TransitionTool,
    fork: Fork,
    eips: Optional[List[int]] = None,
    fixture_format: FixtureFormat,
    **_,
) -> BaseFixture:
    """Generate the BlockchainTest fixture."""
    if fixture_format == EOFFixture:
        return self.make_eof_test_fixture(fork=fork, eips=eips)
    elif fixture_format in StateTest.supported_fixture_formats:
        return self.generate_state_test(fork).generate(
            t8n=t8n, fork=fork, fixture_format=fixture_format, eips=eips
        )
    raise Exception(f"Unknown fixture format: {fixture_format}")

execute(*, fork, execute_format, eips=None)

Generate the list of test fixtures.

Source code in src/ethereum_test_specs/eof.py 
492
493
494
495
496
497
498
499
500
501
502
503
504
def execute(
    self,
    *,
    fork: Fork,
    execute_format: ExecuteFormat,
    eips: Optional[List[int]] = None,
) -> BaseExecute:
    """Generate the list of test fixtures."""
    if execute_format == TransactionPost:
        return self.generate_state_test(fork).execute(
            fork=fork, execute_format=execute_format, eips=eips
        )
    raise Exception(f"Unsupported execute format: {execute_format}")

StateTest

Bases: BaseTest

Filler type that tests transactions over the period of a single block.

Source code in src/ethereum_test_specs/state.py 
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
class StateTest(BaseTest):
    """Filler type that tests transactions over the period of a single block."""

    env: Environment = Field(default_factory=Environment)
    pre: Alloc
    post: Alloc
    tx: Transaction
    block_exception: (
        List[TransactionException | BlockException] | TransactionException | BlockException | None
    ) = None
    engine_api_error_code: Optional[EngineAPIError] = None
    blockchain_test_header_verify: Optional[Header] = None
    blockchain_test_rlp_modifier: Optional[Header] = None
    chain_id: int = 1

    supported_fixture_formats: ClassVar[Sequence[FixtureFormat | LabeledFixtureFormat]] = [
        StateFixture,
    ] + [
        LabeledFixtureFormat(
            fixture_format,
            f"{fixture_format.format_name}_from_state_test",
            f"A {fixture_format.format_name} generated from a state_test",
        )
        for fixture_format in BlockchainTest.supported_fixture_formats
    ]
    supported_execute_formats: ClassVar[Sequence[LabeledExecuteFormat]] = [
        LabeledExecuteFormat(
            TransactionPost,
            "state_test",
            "An execute test derived from a state test",
        ),
    ]

    supported_markers: ClassVar[Dict[str, str]] = {
        "state_test_only": "Only generate a state test fixture",
    }

    @classmethod
    def discard_fixture_format_by_marks(
        cls,
        fixture_format: FixtureFormat,
        fork: Fork,
        markers: List[pytest.Mark],
    ) -> bool:
        """Discard a fixture format from filling if the appropriate marker is used."""
        if "state_test_only" in [m.name for m in markers]:
            return fixture_format != StateFixture
        return False

    def _generate_blockchain_genesis_environment(self, *, fork: Fork) -> Environment:
        """Generate the genesis environment for the BlockchainTest formatted test."""
        assert self.env.number >= 1, (
            "genesis block number cannot be negative, set state test env.number to 1"
        )

        # Modify values to the proper values for the genesis block
        # TODO: All of this can be moved to a new method in `Fork`
        updated_values: Dict[str, Any] = {
            "withdrawals": None,
            "parent_beacon_block_root": None,
            "number": self.env.number - 1,
        }
        if self.env.excess_blob_gas:
            # The excess blob gas environment value means the value of the context (block header)
            # where the transaction is executed. In a blockchain test, we need to indirectly
            # set the excess blob gas by setting the excess blob gas of the genesis block
            # to the expected value plus the TARGET_BLOB_GAS_PER_BLOCK, which is the value
            # that will be subtracted from the excess blob gas when the first block is mined.
            updated_values["excess_blob_gas"] = self.env.excess_blob_gas + (
                fork.target_blobs_per_block() * fork.blob_gas_per_blob()
            )
        if self.env.base_fee_per_gas:
            # Calculate genesis base fee per gas from state test's block#1 env
            updated_values["base_fee_per_gas"] = HexNumber(
                int(int(str(self.env.base_fee_per_gas), 0) * 8 / 7)
            )
        if fork.header_prev_randao_required():
            # Set current random
            updated_values["difficulty"] = None
            updated_values["prev_randao"] = (
                self.env.prev_randao if self.env.prev_randao is not None else self.env.difficulty
            )

        return self.env.copy(**updated_values)

    def _generate_blockchain_blocks(self, *, fork: Fork) -> List[Block]:
        """Generate the single block that represents this state test in a BlockchainTest format."""
        kwargs = {
            "number": self.env.number,
            "timestamp": self.env.timestamp,
            "prev_randao": self.env.prev_randao,
            "fee_recipient": self.env.fee_recipient,
            "gas_limit": self.env.gas_limit,
            "extra_data": self.env.extra_data,
            "withdrawals": self.env.withdrawals,
            "parent_beacon_block_root": self.env.parent_beacon_block_root,
            "txs": [self.tx],
            "ommers": [],
            "header_verify": self.blockchain_test_header_verify,
            "rlp_modifier": self.blockchain_test_rlp_modifier,
        }
        if not fork.header_prev_randao_required():
            kwargs["difficulty"] = self.env.difficulty
        if "block_exception" in self.model_fields_set:
            kwargs["exception"] = self.block_exception  # type: ignore
        elif "error" in self.tx.model_fields_set:
            kwargs["exception"] = self.tx.error  # type: ignore
        return [Block(**kwargs)]

    def generate_blockchain_test(self, *, fork: Fork) -> BlockchainTest:
        """Generate a BlockchainTest fixture from this StateTest fixture."""
        return BlockchainTest.from_test(
            base_test=self,
            genesis_environment=self._generate_blockchain_genesis_environment(fork=fork),
            pre=self.pre,
            post=self.post,
            blocks=self._generate_blockchain_blocks(fork=fork),
        )

    def make_state_test_fixture(
        self,
        t8n: TransitionTool,
        fork: Fork,
        eips: Optional[List[int]] = None,
    ) -> StateFixture:
        """Create a fixture from the state test definition."""
        # We can't generate a state test fixture that names a transition fork,
        # so we get the fork at the block number and timestamp of the state test
        fork = fork.fork_at(self.env.number, self.env.timestamp)

        env = self.env.set_fork_requirements(fork)
        tx = self.tx.with_signature_and_sender(keep_secret_key=True)
        pre_alloc = Alloc.merge(
            Alloc.model_validate(fork.pre_allocation()),
            self.pre,
        )
        if empty_accounts := pre_alloc.empty_accounts():
            raise Exception(f"Empty accounts in pre state: {empty_accounts}")

        transition_tool_output = t8n.evaluate(
            alloc=pre_alloc,
            txs=[tx],
            env=env,
            fork=fork,
            chain_id=self.chain_id,
            reward=0,  # Reward on state tests is always zero
            blob_schedule=fork.blob_schedule(),
            eips=eips,
            debug_output_path=self.get_next_transition_tool_output_path(),
            state_test=True,
            slow_request=self.is_slow_test(),
        )

        try:
            self.post.verify_post_alloc(transition_tool_output.alloc)
        except Exception as e:
            print_traces(t8n.get_traces())
            raise e

        try:
            verify_transactions(
                txs=[tx],
                result=transition_tool_output.result,
                transition_tool_exceptions_reliable=t8n.exception_mapper.reliable,
            )
        except Exception as e:
            print_traces(t8n.get_traces())
            pprint(transition_tool_output.result)
            pprint(transition_tool_output.alloc)
            raise e

        return StateFixture(
            env=FixtureEnvironment(**env.model_dump(exclude_none=True)),
            pre=pre_alloc,
            post={
                fork.blockchain_test_network_name(): [
                    FixtureForkPost(
                        state_root=transition_tool_output.result.state_root,
                        logs_hash=transition_tool_output.result.logs_hash,
                        tx_bytes=tx.rlp(),
                        expect_exception=tx.error,
                        state=transition_tool_output.alloc,
                    )
                ]
            },
            transaction=FixtureTransaction.from_transaction(tx),
            config=FixtureConfig(
                blob_schedule=FixtureBlobSchedule.from_blob_schedule(fork.blob_schedule()),
                chain_id=self.chain_id,
            ),
        )

    def generate(
        self,
        t8n: TransitionTool,
        fork: Fork,
        fixture_format: FixtureFormat,
        eips: Optional[List[int]] = None,
    ) -> BaseFixture:
        """Generate the BlockchainTest fixture."""
        self.check_exception_test(exception=self.tx.error is not None)
        if fixture_format in BlockchainTest.supported_fixture_formats:
            return self.generate_blockchain_test(fork=fork).generate(
                t8n=t8n, fork=fork, fixture_format=fixture_format, eips=eips
            )
        elif fixture_format == StateFixture:
            return self.make_state_test_fixture(t8n, fork, eips)

        raise Exception(f"Unknown fixture format: {fixture_format}")

    def execute(
        self,
        *,
        fork: Fork,
        execute_format: ExecuteFormat,
        eips: Optional[List[int]] = None,
    ) -> BaseExecute:
        """Generate the list of test fixtures."""
        if execute_format == TransactionPost:
            return TransactionPost(
                blocks=[[self.tx]],
                post=self.post,
            )
        raise Exception(f"Unsupported execute format: {execute_format}")

discard_fixture_format_by_marks(fixture_format, fork, markers) classmethod

Discard a fixture format from filling if the appropriate marker is used.

Source code in src/ethereum_test_specs/state.py 
77
78
79
80
81
82
83
84
85
86
87
@classmethod
def discard_fixture_format_by_marks(
    cls,
    fixture_format: FixtureFormat,
    fork: Fork,
    markers: List[pytest.Mark],
) -> bool:
    """Discard a fixture format from filling if the appropriate marker is used."""
    if "state_test_only" in [m.name for m in markers]:
        return fixture_format != StateFixture
    return False

generate_blockchain_test(*, fork)

Generate a BlockchainTest fixture from this StateTest fixture.

Source code in src/ethereum_test_specs/state.py 
149
150
151
152
153
154
155
156
157
def generate_blockchain_test(self, *, fork: Fork) -> BlockchainTest:
    """Generate a BlockchainTest fixture from this StateTest fixture."""
    return BlockchainTest.from_test(
        base_test=self,
        genesis_environment=self._generate_blockchain_genesis_environment(fork=fork),
        pre=self.pre,
        post=self.post,
        blocks=self._generate_blockchain_blocks(fork=fork),
    )

make_state_test_fixture(t8n, fork, eips=None)

Create a fixture from the state test definition.

Source code in src/ethereum_test_specs/state.py 
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
def make_state_test_fixture(
    self,
    t8n: TransitionTool,
    fork: Fork,
    eips: Optional[List[int]] = None,
) -> StateFixture:
    """Create a fixture from the state test definition."""
    # We can't generate a state test fixture that names a transition fork,
    # so we get the fork at the block number and timestamp of the state test
    fork = fork.fork_at(self.env.number, self.env.timestamp)

    env = self.env.set_fork_requirements(fork)
    tx = self.tx.with_signature_and_sender(keep_secret_key=True)
    pre_alloc = Alloc.merge(
        Alloc.model_validate(fork.pre_allocation()),
        self.pre,
    )
    if empty_accounts := pre_alloc.empty_accounts():
        raise Exception(f"Empty accounts in pre state: {empty_accounts}")

    transition_tool_output = t8n.evaluate(
        alloc=pre_alloc,
        txs=[tx],
        env=env,
        fork=fork,
        chain_id=self.chain_id,
        reward=0,  # Reward on state tests is always zero
        blob_schedule=fork.blob_schedule(),
        eips=eips,
        debug_output_path=self.get_next_transition_tool_output_path(),
        state_test=True,
        slow_request=self.is_slow_test(),
    )

    try:
        self.post.verify_post_alloc(transition_tool_output.alloc)
    except Exception as e:
        print_traces(t8n.get_traces())
        raise e

    try:
        verify_transactions(
            txs=[tx],
            result=transition_tool_output.result,
            transition_tool_exceptions_reliable=t8n.exception_mapper.reliable,
        )
    except Exception as e:
        print_traces(t8n.get_traces())
        pprint(transition_tool_output.result)
        pprint(transition_tool_output.alloc)
        raise e

    return StateFixture(
        env=FixtureEnvironment(**env.model_dump(exclude_none=True)),
        pre=pre_alloc,
        post={
            fork.blockchain_test_network_name(): [
                FixtureForkPost(
                    state_root=transition_tool_output.result.state_root,
                    logs_hash=transition_tool_output.result.logs_hash,
                    tx_bytes=tx.rlp(),
                    expect_exception=tx.error,
                    state=transition_tool_output.alloc,
                )
            ]
        },
        transaction=FixtureTransaction.from_transaction(tx),
        config=FixtureConfig(
            blob_schedule=FixtureBlobSchedule.from_blob_schedule(fork.blob_schedule()),
            chain_id=self.chain_id,
        ),
    )

generate(t8n, fork, fixture_format, eips=None)

Generate the BlockchainTest fixture.

Source code in src/ethereum_test_specs/state.py 
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
def generate(
    self,
    t8n: TransitionTool,
    fork: Fork,
    fixture_format: FixtureFormat,
    eips: Optional[List[int]] = None,
) -> BaseFixture:
    """Generate the BlockchainTest fixture."""
    self.check_exception_test(exception=self.tx.error is not None)
    if fixture_format in BlockchainTest.supported_fixture_formats:
        return self.generate_blockchain_test(fork=fork).generate(
            t8n=t8n, fork=fork, fixture_format=fixture_format, eips=eips
        )
    elif fixture_format == StateFixture:
        return self.make_state_test_fixture(t8n, fork, eips)

    raise Exception(f"Unknown fixture format: {fixture_format}")

execute(*, fork, execute_format, eips=None)

Generate the list of test fixtures.

Source code in src/ethereum_test_specs/state.py 
250
251
252
253
254
255
256
257
258
259
260
261
262
263
def execute(
    self,
    *,
    fork: Fork,
    execute_format: ExecuteFormat,
    eips: Optional[List[int]] = None,
) -> BaseExecute:
    """Generate the list of test fixtures."""
    if execute_format == TransactionPost:
        return TransactionPost(
            blocks=[[self.tx]],
            post=self.post,
        )
    raise Exception(f"Unsupported execute format: {execute_format}")

StateStaticTest

Bases: StateTestInFiller, BaseStaticTest

General State Test static filler from ethereum/tests.

Source code in src/ethereum_test_specs/static_state/state_static.py 
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
class StateStaticTest(StateTestInFiller, BaseStaticTest):
    """General State Test static filler from ethereum/tests."""

    test_name: str = ""
    vectors: List[StateTestVector] | None = None
    format_name: ClassVar[str] = "state_test"

    def model_post_init(self, context):
        """Initialize StateStaticTest."""
        super().model_post_init(context)

    def fill_function(self) -> Callable:
        """Return a StateTest spec from a static file."""
        d_g_v_parameters: List[ParameterSet] = []
        for d in range(len(self.transaction.data)):
            for g in range(len(self.transaction.gas_limit)):
                for v in range(len(self.transaction.value)):
                    exception_test = False
                    for expect in self.expect:
                        if expect.has_index(d, g, v) and expect.expect_exception is not None:
                            exception_test = True
                    # TODO: This does not take into account exceptions that only happen on
                    #       specific forks, but this requires a covariant parametrize
                    marks = [pytest.mark.exception_test] if exception_test else []
                    d_g_v_parameters.append(
                        pytest.param(d, g, v, marks=marks, id=f"d{d}-g{g}-v{v}")
                    )

        @pytest.mark.valid_at(*self.get_valid_at_forks())
        @pytest.mark.parametrize("d,g,v", d_g_v_parameters)
        def test_state_vectors(
            state_test: StateTestFiller,
            fork: Fork,
            d: int,
            g: int,
            v: int,
        ):
            for expect in self.expect:
                if expect.has_index(d, g, v):
                    if fork.name() in expect.network:
                        post, tx = self._make_vector(
                            d,
                            g,
                            v,
                            expect.result,
                            exception=None
                            if expect.expect_exception is None
                            else expect.expect_exception[fork.name()],
                        )
                        return state_test(
                            env=self._get_env,
                            pre=self._get_pre,
                            post=post,
                            tx=tx,
                        )
            pytest.fail(f"Expectation not found for d={d}, g={g}, v={v}, fork={fork}")

        if self.info and self.info.pytest_marks:
            for mark in self.info.pytest_marks:
                apply_mark = getattr(pytest.mark, mark)
                test_state_vectors = apply_mark(test_state_vectors)

        return test_state_vectors

    def get_valid_at_forks(self) -> List[str]:
        """Return list of forks that are valid for this test."""
        fork_list: List[str] = []
        for expect in self.expect:
            for fork in expect.network:
                if fork not in fork_list:
                    fork_list.append(fork)
        return fork_list

    @cached_property
    def _get_env(self) -> Environment:
        """Parse environment."""
        # Convert Environment data from .json filler into pyspec type
        test_env = self.env
        env = Environment(
            fee_recipient=Address(test_env.current_coinbase),
            difficulty=ZeroPaddedHexNumber(test_env.current_difficulty)
            if test_env.current_difficulty is not None
            else None,
            prev_randao=ZeroPaddedHexNumber(test_env.current_random)
            if test_env.current_random is not None
            else None,
            gas_limit=ZeroPaddedHexNumber(test_env.current_gas_limit),
            number=ZeroPaddedHexNumber(test_env.current_number),
            timestamp=ZeroPaddedHexNumber(test_env.current_timestamp),
            base_fee_per_gas=ZeroPaddedHexNumber(test_env.current_base_fee)
            if test_env.current_base_fee is not None
            else None,
            excess_blob_gas=ZeroPaddedHexNumber(test_env.current_excess_blob_gas)
            if test_env.current_excess_blob_gas is not None
            else None,
        )
        return env

    @cached_property
    def _get_pre(self) -> Alloc:
        """Parse pre."""
        # Convert pre state data from .json filler into pyspec type
        pre = Alloc()
        for account_address, account in self.pre.items():
            storage: Storage = Storage()
            for key, value in account.storage.items():
                storage[key] = value

            pre[account_address] = Account(
                balance=account.balance,
                nonce=account.nonce,
                code=account.code.compiled,
                storage=storage,
            )
        return pre

    def _make_vector(
        self,
        d: int,
        g: int,
        v: int,
        expect_result: Dict[AddressInFiller, AccountInExpectSection],
        exception: TransactionExceptionInstanceOrList | None,
    ) -> Tuple[Alloc, Transaction]:
        """Compose test vector from test data."""
        general_tr = self.transaction
        data_box = general_tr.data[d]

        tr: Transaction = Transaction(
            data=data_box.data.compiled,
            access_list=data_box.access_list,
            gas_limit=HexNumber(general_tr.gas_limit[g]),
            value=HexNumber(general_tr.value[v]),
            gas_price=general_tr.gas_price,
            max_fee_per_gas=general_tr.max_fee_per_gas,
            max_priority_fee_per_gas=general_tr.max_priority_fee_per_gas,
            max_fee_per_blob_gas=general_tr.max_fee_per_blob_gas,
            blob_versioned_hashes=general_tr.blob_versioned_hashes,
            nonce=HexNumber(general_tr.nonce),
            to=Address(general_tr.to) if general_tr.to is not None else None,
            secret_key=Hash(general_tr.secret_key),
            error=exception,
        )

        post = Alloc()
        for address, account in expect_result.items():
            if account.expected_to_not_exist is not None:
                post[address] = Account.NONEXISTENT
                continue

            account_kwargs: Dict[str, Any] = {}
            if account.storage is not None:
                storage = Storage()
                for key, value in account.storage.items():
                    if value != "ANY":
                        storage[key] = value
                    else:
                        storage.set_expect_any(key)
                account_kwargs["storage"] = storage
            if account.code is not None:
                account_kwargs["code"] = account.code.compiled
            if account.balance is not None:
                account_kwargs["balance"] = account.balance
            if account.nonce is not None:
                account_kwargs["nonce"] = account.nonce

            post[address] = Account(**account_kwargs)

        return post, tr

model_post_init(context)

Initialize StateStaticTest.

Source code in src/ethereum_test_specs/static_state/state_static.py 
30
31
32
def model_post_init(self, context):
    """Initialize StateStaticTest."""
    super().model_post_init(context)

fill_function()

Return a StateTest spec from a static file.

Source code in src/ethereum_test_specs/static_state/state_static.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
def fill_function(self) -> Callable:
    """Return a StateTest spec from a static file."""
    d_g_v_parameters: List[ParameterSet] = []
    for d in range(len(self.transaction.data)):
        for g in range(len(self.transaction.gas_limit)):
            for v in range(len(self.transaction.value)):
                exception_test = False
                for expect in self.expect:
                    if expect.has_index(d, g, v) and expect.expect_exception is not None:
                        exception_test = True
                # TODO: This does not take into account exceptions that only happen on
                #       specific forks, but this requires a covariant parametrize
                marks = [pytest.mark.exception_test] if exception_test else []
                d_g_v_parameters.append(
                    pytest.param(d, g, v, marks=marks, id=f"d{d}-g{g}-v{v}")
                )

    @pytest.mark.valid_at(*self.get_valid_at_forks())
    @pytest.mark.parametrize("d,g,v", d_g_v_parameters)
    def test_state_vectors(
        state_test: StateTestFiller,
        fork: Fork,
        d: int,
        g: int,
        v: int,
    ):
        for expect in self.expect:
            if expect.has_index(d, g, v):
                if fork.name() in expect.network:
                    post, tx = self._make_vector(
                        d,
                        g,
                        v,
                        expect.result,
                        exception=None
                        if expect.expect_exception is None
                        else expect.expect_exception[fork.name()],
                    )
                    return state_test(
                        env=self._get_env,
                        pre=self._get_pre,
                        post=post,
                        tx=tx,
                    )
        pytest.fail(f"Expectation not found for d={d}, g={g}, v={v}, fork={fork}")

    if self.info and self.info.pytest_marks:
        for mark in self.info.pytest_marks:
            apply_mark = getattr(pytest.mark, mark)
            test_state_vectors = apply_mark(test_state_vectors)

    return test_state_vectors

get_valid_at_forks()

Return list of forks that are valid for this test.

Source code in src/ethereum_test_specs/static_state/state_static.py 
87
88
89
90
91
92
93
94
def get_valid_at_forks(self) -> List[str]:
    """Return list of forks that are valid for this test."""
    fork_list: List[str] = []
    for expect in self.expect:
        for fork in expect.network:
            if fork not in fork_list:
                fork_list.append(fork)
    return fork_list

TransactionTest

Bases: BaseTest

Filler type that tests the transaction over the period of a single block.

Source code in src/ethereum_test_specs/transaction.py 
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
class TransactionTest(BaseTest):
    """Filler type that tests the transaction over the period of a single block."""

    tx: Transaction
    pre: Alloc | None = None

    supported_fixture_formats: ClassVar[Sequence[FixtureFormat | LabeledFixtureFormat]] = [
        TransactionFixture,
    ]
    supported_execute_formats: ClassVar[Sequence[LabeledExecuteFormat]] = [
        LabeledExecuteFormat(
            TransactionPost,
            "transaction_test",
            "An execute test derived from a transaction test",
        ),
    ]

    def make_transaction_test_fixture(
        self,
        fork: Fork,
        eips: Optional[List[int]] = None,
    ) -> TransactionFixture:
        """Create a fixture from the transaction test definition."""
        if self.tx.error is not None:
            result = FixtureResult(
                exception=self.tx.error,
                hash=None,
                intrinsic_gas=0,
                sender=None,
            )
        else:
            intrinsic_gas_cost_calculator = fork.transaction_intrinsic_cost_calculator()
            intrinsic_gas = intrinsic_gas_cost_calculator(
                calldata=self.tx.data,
                contract_creation=self.tx.to is None,
                access_list=self.tx.access_list,
                authorization_list_or_count=self.tx.authorization_list,
            )
            result = FixtureResult(
                exception=None,
                hash=self.tx.hash,
                intrinsic_gas=intrinsic_gas,
                sender=self.tx.sender,
            )

        return TransactionFixture(
            result={
                fork.blockchain_test_network_name(): result,
            },
            transaction=self.tx.with_signature_and_sender().rlp(),
        )

    def generate(
        self,
        t8n: TransitionTool,
        fork: Fork,
        fixture_format: FixtureFormat,
        eips: Optional[List[int]] = None,
    ) -> BaseFixture:
        """Generate the TransactionTest fixture."""
        self.check_exception_test(exception=self.tx.error is not None)
        if fixture_format == TransactionFixture:
            return self.make_transaction_test_fixture(fork, eips)

        raise Exception(f"Unknown fixture format: {fixture_format}")

    def execute(
        self,
        *,
        fork: Fork,
        execute_format: ExecuteFormat,
        eips: Optional[List[int]] = None,
    ) -> BaseExecute:
        """Execute the transaction test by sending it to the live network."""
        if execute_format == TransactionPost:
            return TransactionPost(
                blocks=[[self.tx]],
                post={},
            )
        raise Exception(f"Unsupported execute format: {execute_format}")

make_transaction_test_fixture(fork, eips=None)

Create a fixture from the transaction test definition.

Source code in src/ethereum_test_specs/transaction.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
def make_transaction_test_fixture(
    self,
    fork: Fork,
    eips: Optional[List[int]] = None,
) -> TransactionFixture:
    """Create a fixture from the transaction test definition."""
    if self.tx.error is not None:
        result = FixtureResult(
            exception=self.tx.error,
            hash=None,
            intrinsic_gas=0,
            sender=None,
        )
    else:
        intrinsic_gas_cost_calculator = fork.transaction_intrinsic_cost_calculator()
        intrinsic_gas = intrinsic_gas_cost_calculator(
            calldata=self.tx.data,
            contract_creation=self.tx.to is None,
            access_list=self.tx.access_list,
            authorization_list_or_count=self.tx.authorization_list,
        )
        result = FixtureResult(
            exception=None,
            hash=self.tx.hash,
            intrinsic_gas=intrinsic_gas,
            sender=self.tx.sender,
        )

    return TransactionFixture(
        result={
            fork.blockchain_test_network_name(): result,
        },
        transaction=self.tx.with_signature_and_sender().rlp(),
    )

generate(t8n, fork, fixture_format, eips=None)

Generate the TransactionTest fixture.

Source code in src/ethereum_test_specs/transaction.py 
77
78
79
80
81
82
83
84
85
86
87
88
89
def generate(
    self,
    t8n: TransitionTool,
    fork: Fork,
    fixture_format: FixtureFormat,
    eips: Optional[List[int]] = None,
) -> BaseFixture:
    """Generate the TransactionTest fixture."""
    self.check_exception_test(exception=self.tx.error is not None)
    if fixture_format == TransactionFixture:
        return self.make_transaction_test_fixture(fork, eips)

    raise Exception(f"Unknown fixture format: {fixture_format}")

execute(*, fork, execute_format, eips=None)

Execute the transaction test by sending it to the live network.

Source code in src/ethereum_test_specs/transaction.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def execute(
    self,
    *,
    fork: Fork,
    execute_format: ExecuteFormat,
    eips: Optional[List[int]] = None,
) -> BaseExecute:
    """Execute the transaction test by sending it to the live network."""
    if execute_format == TransactionPost:
        return TransactionPost(
            blocks=[[self.tx]],
            post={},
        )
    raise Exception(f"Unsupported execute format: {execute_format}")