-
Notifications
You must be signed in to change notification settings - Fork 25
/
Copy pathconftest_ganache.py
135 lines (103 loc) · 3.49 KB
/
conftest_ganache.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
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
#
# Copyright 2024 Ocean Protocol Foundation
# SPDX-License-Identifier: Apache-2.0
#
from unittest.mock import Mock
import pytest
from pdr_backend.cli.arg_feed import ArgFeed
from pdr_backend.contract.pred_submitter_mgr import PredSubmitterMgr
from pdr_backend.contract.predictoor_batcher import PredictoorBatcher
from pdr_backend.contract.feed_contract import FeedContract
from pdr_backend.contract.token import Token
from pdr_backend.ppss.ppss import PPSS, fast_test_yaml_str
from pdr_backend.pred_submitter.deploy import deploy_pred_submitter_mgr_contract
from pdr_backend.publisher.publish_asset import publish_asset
from pdr_backend.util.currency_types import Eth
CHAIN_ID = 8996
S_PER_EPOCH = 300
@pytest.fixture(scope="session") # "session" = invoke once, across all tests
def chain_id():
return CHAIN_ID
@pytest.fixture(scope="session")
def web3_config():
return _web3_config()
def _web3_config():
return _web3_pp().web3_config
@pytest.fixture(scope="session")
def rpc_url():
return _web3_pp().rpc_url
@pytest.fixture(scope="session")
def web3_pp():
return _web3_pp()
def _web3_pp():
return _ppss().web3_pp
@pytest.fixture(scope="session")
def ppss():
return _ppss()
def _ppss():
s = fast_test_yaml_str()
return PPSS(yaml_str=s, network="development")
@pytest.fixture(scope="session")
def OCEAN() -> Token:
token_address = _web3_pp().get_address("Ocean")
return Token(_web3_pp(), token_address)
@pytest.fixture(scope="module") # "module" = invoke once per test module
def feed_contract1():
w3p = _web3_pp()
w3c = w3p.web3_config
_, _, _, _, logs = publish_asset(
s_per_epoch=S_PER_EPOCH,
s_per_subscription=S_PER_EPOCH * 24,
feed=ArgFeed("kraken", "close", "ETH/USDT", "5m"),
trueval_submitter_addr=w3c.owner,
feeCollector_addr=w3c.owner,
rate=Eth(3),
cut=Eth(0.2),
web3_pp=w3p,
)
dt_addr = logs["newTokenAddress"]
return FeedContract(w3p, dt_addr)
@pytest.fixture(scope="module")
def feed_contract2():
w3p = _web3_pp()
w3c = w3p.web3_config
_, _, _, _, logs = publish_asset(
s_per_epoch=S_PER_EPOCH,
s_per_subscription=S_PER_EPOCH * 24,
feed=ArgFeed("kraken", "close", "ETH/USDT", "5m"),
trueval_submitter_addr=w3c.owner,
feeCollector_addr=w3c.owner,
rate=Eth(3),
cut=Eth(0.2),
web3_pp=w3p,
)
dt_addr = logs["newTokenAddress"]
return FeedContract(w3p, dt_addr)
@pytest.fixture(scope="module") # "module" = invoke once per test module
def feed_contract_empty():
w3p = _web3_pp()
w3c = w3p.web3_config
_, _, _, _, logs = publish_asset(
s_per_epoch=S_PER_EPOCH,
s_per_subscription=S_PER_EPOCH * 24,
feed=ArgFeed("kraken", "close", "ETH/USDT", "5m"),
trueval_submitter_addr=w3c.owner,
feeCollector_addr=w3c.owner,
rate=Eth(3),
cut=Eth(0.2),
web3_pp=w3p,
)
dt_addr = logs["newTokenAddress"]
predictoor_c = FeedContract(w3p, dt_addr)
predictoor_c.get_exchanges = Mock(return_value=[])
return predictoor_c
# pylint: disable=redefined-outer-name
@pytest.fixture(scope="module")
def predictoor_batcher():
w3p = _web3_pp()
predictoor_batcher_addr = w3p.get_address("PredictoorHelper")
return PredictoorBatcher(w3p, predictoor_batcher_addr)
@pytest.fixture()
def pred_submitter_mgr(web3_pp):
contract_address = deploy_pred_submitter_mgr_contract(web3_pp)
return PredSubmitterMgr(web3_pp, contract_address)