conftest.py

importjson
importwarnings
frompathlibimportPath

importpytest
frombrownie._configimportCONFIG
frombrownie.project.mainimportget_loaded_projects

# functions in wrapped methods are renamed to simplify common tests

WRAPPED_COIN_METHODS= {
"ATokenMock": {"get_rate": "_get_rate", "mint": "mint"},
"cERC20": {"get_rate": "exchangeRateStored", "mint": "mint"},
"IdleToken": {"get_rate": "tokenPrice", "mint": "mintIdleToken"},
"renERC20": {"get_rate": "exchangeRateCurrent"},
"yERC20": {"get_rate": "getPricePerFullShare", "mint": "deposit"},
"aETH": {"get_rate": "ratio"},
"rETH": {"get_rate": "getExchangeRate"},
}

pytest_plugins= [
"fixtures.accounts",
"fixtures.coins",
"fixtures.deployments",
"fixtures.functions",
"fixtures.pooldata",
"fixtures.setup",
]

_pooldata= {}


defpytest_addoption(parser):
parser.addoption("--pool", help="comma-separated list of pools to target")
parser.addoption("--unitary", action="store_true", help="only run unit tests")
parser.addoption("--integration", action="store_true", help="only run integration tests")


defpytest_configure(config):
# add custom markers
config.addinivalue_line("markers", "target_pool: run test against one or more specific pool")
config.addinivalue_line("markers", "skip_pool: exclude one or more pools in this test")
config.addinivalue_line(
"markers", "skip_pool_type: exclude one or more pool types in this test"
 )
config.addinivalue_line("markers", "lending: only run test against pools that involve lending")
config.addinivalue_line("markers", "zap: only run test against pools with a deposit contract")
config.addinivalue_line(
"markers",
"itercoins: parametrize a test with one or more ranges, equal to the length "
"of `wrapped_coins` for the active pool",
 )


defpytest_sessionstart():
# load `pooldata.json` for each pool
project=get_loaded_projects()[0]
forpathin [iforiinproject._path.glob("contracts/pools/*") ifi.is_dir()]:
withpath.joinpath("pooldata.json").open() asfp:
_pooldata[path.name] =json.load(fp)
_pooldata[path.name].update(
name=path.name, swap_contract=next(i.stemforiinpath.glob("StableSwap*"))
 )
zap_contract=next((i.stemforiinpath.glob("Deposit*")), None)
ifzap_contract:
_pooldata[path.name]["zap_contract"] =zap_contract

# create pooldata for templates
lp_contract=sorted(i._nameforiinprojectifi._name.startswith("CurveToken"))[-1]

forpathin [iforiinproject._path.glob("contracts/pool-templates/*") ifi.is_dir()]:
withpath.joinpath("pooldata.json").open() asfp:
name=f"template-{path.name}"
_pooldata[name] =json.load(fp)
_pooldata[name].update(
name=name,
lp_contract=lp_contract,
swap_contract=next(i.stemforiinpath.glob("*Swap*")),
 )
zap_contract=next((i.stemforiinpath.glob("Deposit*")), None)
ifzap_contract:
_pooldata[name]["zap_contract"] =zap_contract

for_, datain_pooldata.items():
if"base_pool"indata:
data["base_pool"] =_pooldata[data["base_pool"]]
elif"base_pool_contract"indata:
# for metapool templates, we target a contract instead of a specific pool
base_swap=data["base_pool_contract"]
base_data=next(vforvin_pooldata.values() ifv["swap_contract"] ==base_swap)
data["base_pool"] =base_data


defpytest_ignore_collect(path, config):
project=get_loaded_projects()[0]
path=Path(path).relative_to(project._path)
path_parts=path.parts[1:-1]

ifpath.is_dir():
returnNone

# always collect fixtures
ifpath_parts[:1] == ("fixtures",):
returnNone

# always allow forked tests
ifpath_parts[:1] == ("forked",):
returnNone

# with the `--unitary` flag, skip any tests in an `integration` subdirectory
ifconfig.getoption("unitary") and"integration"inpath_parts:
returnTrue

# with the `--integration` flag, skip any tests NOT in an `integration` subdirectory
ifconfig.getoption("integration") and"integration"notinpath_parts:
returnTrue

ifconfig.getoption("pool") andpath_parts:
# with a specific pool targeted, only run pool and zap tests
ifpath_parts[0] notin ("pools", "zaps"):
returnTrue

# always run common tests
ifpath_parts[1] =="common":
returnNone

target_pools=config.getoption("pool").split(",")

# only include metapool tests if at least one targeted pool is a metapool
ifpath_parts[1] =="meta":
returnnext(
 (Noneforiintarget_poolsif"meta"in_pooldata[i].get("pool_types", [])), True
 )

# only include a-style tests if at least one targeted pool is an a-style pool
ifpath_parts[1] =="arate":
returnnext(
 (Noneforiintarget_poolsif"arate"in_pooldata[i].get("pool_types", [])), True
 )

# only include c-style tests if at least one targeted pool is an c-style pool
ifpath_parts[1] =="crate":
returnnext(
 (Noneforiintarget_poolsif"crate"in_pooldata[i].get("pool_type", [])), True
 )

# only include eth tests if at least one targeted pool is an eth pool
ifpath_parts[1] =="eth":
returnnext(
 (Noneforiintarget_poolsif"eth"in_pooldata[i].get("pool_type", [])), True
 )

# filter other pool-specific folders
ifpath_parts[1] notintarget_pools:
returnTrue


defpytest_generate_tests(metafunc):
project=get_loaded_projects()[0]
itercoins_bound=max(len(i["coins"]) foriin_pooldata.values())
if"pool_data"inmetafunc.fixturenames:
# parametrize `pool_data`
test_path=Path(metafunc.definition.fspath).relative_to(project._path)
iftest_path.parts[1] in ("pools", "zaps"):
iftest_path.parts[2] in ("common", "meta", "crate", "arate", "eth"):
# parametrize common pool/zap tests to run against all pools
ifmetafunc.config.getoption("pool"):
params=metafunc.config.getoption("pool").split(",")
else:
params=list(_pooldata)
# parameterize based on pool type
iftest_path.parts[2] =="meta":
params= [iforiinparamsif"meta"in_pooldata[i].get("pool_types", [])]
iftest_path.parts[2] =="arate":
params= [iforiinparamsif"arate"in_pooldata[i].get("pool_types", [])]
iftest_path.parts[2] =="crate":
params= [iforiinparamsif"crate"in_pooldata[i].get("pool_types", [])]
iftest_path.parts[2] =="eth":
params= [iforiinparamsif"eth"in_pooldata[i].get("pool_types", [])]

else:
# run targetted pool/zap tests against only the specific pool
params= [test_path.parts[2]]

iftest_path.parts[1] =="zaps":
# for zap tests, filter by pools that have a Deposit contract
params= [iforiinparamsif_pooldata[i].get("zap_contract")]
else:
# pool tests outside `tests/pools` or `tests/zaps` will only run when
# a target pool is explicitly declared
try:
params=metafunc.config.getoption("pool").split(",")
exceptException:
params= []
warnings.warn(
f"'{test_path.as_posix()}' contains pool tests, but is outside of "
"'tests/pools/'. To run it, specify a pool with `--pool [name]`"
 )
metafunc.parametrize("pool_data", params, indirect=True, scope="session")

# apply initial parametrization of `itercoins`
formarkerinmetafunc.definition.iter_markers(name="itercoins"):
foriteminmarker.args:
metafunc.parametrize(item, range(itercoins_bound))


defpytest_collection_modifyitems(config, items):
project=get_loaded_projects()[0]
try:
is_forked="fork"inCONFIG.active_network["id"]
exceptException:
is_forked=False

foriteminitems.copy():
try:
params=item.callspec.params
data=_pooldata[params["pool_data"]]
exceptException:
continue

# during forked tests, filter pools where pooldata does not contain deployment addresses
ifis_forkedandnext((iforiindata["coins"] if"underlying_address"notini), False):
items.remove(item)
continue

# remove excess `itercoins` parametrized tests
formarkerinitem.iter_markers(name="itercoins"):
n_coins=len(data["coins"])

# for metapools, consider the base pool when calculating n_coins
ifmarker.kwargs.get("underlying") and"base_pool"indata:
n_coins=len(data["base_pool"]["coins"]) +1

values= [params[i] foriinmarker.args]
ifmax(values) >=n_coinsorlen(set(values)) <len(values):
items.remove(item)
break

ifitemnotinitems:
continue

# apply `skip_pool` marker
formarkerinitem.iter_markers(name="skip_pool"):
ifparams["pool_data"] inmarker.args:
items.remove(item)
break

ifitemnotinitems:
continue

# apply `skip_pool_type` marker
formarkerinitem.iter_markers(name="skip_pool_type"):
iflen(set(data.get("pool_types", [])) &set(marker.args)):
items.remove(item)
break

ifitemnotinitems:
continue

# apply `target_pool` marker
formarkerinitem.iter_markers(name="target_pool"):
ifparams["pool_data"] notinmarker.args:
items.remove(item)
break

ifitemnotinitems:
continue

# apply `lending` marker
ifnext(item.iter_markers(name="lending"), False):
deployer=getattr(project, data["swap_contract"])
if"exchange_underlying"notindeployer.signatures:
items.remove(item)
continue

# apply `zap` marker
ifnext(item.iter_markers(name="zap"), False) and"zap_contract"notindata:
items.remove(item)
continue

# hacky magic to ensure the correct number of tests is shown in collection report
config.pluginmanager.get_plugin("terminalreporter")._numcollected=len(items)


@pytest.hookimpl(trylast=True)
defpytest_sessionfinish(session, exitstatus):
ifexitstatus==pytest.ExitCode.NO_TESTS_COLLECTED:
# because of how tests are filtered in the CI, we treat "no tests collected" as passing
session.exitstatus=pytest.ExitCode.OK


# isolation setup


@pytest.fixture(autouse=True)
defisolation_setup(fn_isolation):
pass


# main parametrized fixture, used to pass data about each pool into the other fixtures


@pytest.fixture(scope="module")
defpool_data(request):
project=get_loaded_projects()[0]

ifhasattr(request, "param"):
pool_name=request.param
else:
test_path=Path(request.fspath).relative_to(project._path)
# ("tests", "pools" or "zaps", pool_name, ...)
pool_name=test_path.parts[2]
return_pooldata[pool_name]


@pytest.fixture(scope="module")
defbase_pool_data(pool_data):
returnpool_data.get("base_pool", None)


@pytest.fixture(scope="session")
defproject():
yieldget_loaded_projects()[0]


@pytest.fixture(scope="session")
defis_forked():
yield"fork"inCONFIG.active_network["id"]