56 lines
2.1 KiB
Python
56 lines
2.1 KiB
Python
import os
|
|
from enum import Enum, auto
|
|
from time import sleep
|
|
|
|
import pytest
|
|
import yaml
|
|
|
|
from config import NamedTunnelConfig, QuickTunnelConfig
|
|
from constants import BACKOFF_SECS
|
|
from util import LOGGER
|
|
|
|
|
|
class CfdModes(Enum):
|
|
NAMED = auto()
|
|
QUICK = auto()
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def component_tests_config():
|
|
config_file = os.getenv("COMPONENT_TESTS_CONFIG")
|
|
if config_file is None:
|
|
raise Exception(
|
|
"Need to provide path to config file in COMPONENT_TESTS_CONFIG")
|
|
with open(config_file, 'r') as stream:
|
|
config = yaml.safe_load(stream)
|
|
LOGGER.info(f"component tests base config {config}")
|
|
|
|
def _component_tests_config(additional_config={}, cfd_mode=CfdModes.NAMED, provide_ingress=True):
|
|
# Allows the ingress rules to be omitted from the provided config
|
|
ingress = []
|
|
if provide_ingress:
|
|
ingress = config['ingress']
|
|
|
|
# Provide the hostname to allow routing to the tunnel even if the ingress rule isn't defined in the config
|
|
hostname = config['ingress'][0]['hostname']
|
|
|
|
if cfd_mode is CfdModes.NAMED:
|
|
return NamedTunnelConfig(additional_config=additional_config,
|
|
cloudflared_binary=config['cloudflared_binary'],
|
|
tunnel=config['tunnel'],
|
|
credentials_file=config['credentials_file'],
|
|
ingress=ingress,
|
|
hostname=hostname)
|
|
elif cfd_mode is CfdModes.QUICK:
|
|
return QuickTunnelConfig(additional_config=additional_config, cloudflared_binary=config['cloudflared_binary'])
|
|
else:
|
|
raise Exception(f"Unknown cloudflared mode {cfd_mode}")
|
|
|
|
return _component_tests_config
|
|
|
|
|
|
# This fixture is automatically called before each tests to make sure the previous cloudflared has been shutdown
|
|
@pytest.fixture(autouse=True)
|
|
def wait_previous_cloudflared():
|
|
sleep(BACKOFF_SECS)
|