Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 3 additions & 2 deletions cardano_node_tests/utils/helpers.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,4 @@
import argparse
import collections
import contextlib
import datetime
import functools
Expand All @@ -17,6 +16,7 @@
import subprocess
import types as tt
import typing as tp
from collections import abc

import cardano_node_tests.utils.types as ttypes

Expand Down Expand Up @@ -291,9 +291,10 @@ def tool_has(command: str) -> bool:

def flatten(
iterable: tp.Iterable,
ltypes: type[tp.Iterable[tp.Any]] = collections.abc.Iterable, # pyright: ignore [reportAttributeAccessIssue]
ltypes: type[tp.Iterable] | None = None,
) -> tp.Generator:
"""Flatten an irregular (arbitrarily nested) iterable of iterables."""
ltypes = ltypes if ltypes is not None else abc.Iterable
remainder = iter(iterable)
while True:
try:
Expand Down
5 changes: 3 additions & 2 deletions cardano_node_tests/utils/smash_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
import typing as tp

import requests
from requests import auth as rauth

from cardano_node_tests.utils import cluster_nodes
from cardano_node_tests.utils import dbsync_queries
Expand Down Expand Up @@ -55,11 +56,11 @@ def __init__(self, instance_num: int) -> None:
self.base_url = f"http://localhost:{self.port}"
self.auth = self._get_auth()

def _get_auth(self) -> requests.auth.HTTPBasicAuth | None: # pyright: ignore [reportAttributeAccessIssue]
def _get_auth(self) -> rauth.HTTPBasicAuth | None:
"""Get Basic Auth credentials if configured."""
admin = os.environ.get("SMASH_ADMIN", "admin")
password = os.environ.get("SMASH_PASSWORD", "password")
return requests.auth.HTTPBasicAuth(admin, password) if admin and password else None # pyright: ignore [reportAttributeAccessIssue]
return rauth.HTTPBasicAuth(admin, password) if admin and password else None

def get_pool_metadata(self, pool_id: str, pool_meta_hash: str) -> PoolMetadata:
"""Fetch stake pool metadata from SMASH, returning a `PoolMetadata`."""
Expand Down