cloudflare/cloudflared
Publicmirrored from https://github.com/cloudflare/cloudflaredAvailable
component-tests/config.py
132lines · modecode
| 1 | #!/usr/bin/env python |
| 2 | import base64 |
| 3 | import copy |
| 4 | import os |
| 5 | import yaml |
| 6 | |
| 7 | from dataclasses import dataclass, InitVar |
| 8 | |
| 9 | from constants import METRICS_PORT |
| 10 | |
| 11 | from util import LOGGER |
| 12 | |
| 13 | # frozen=True raises exception when assigning to fields. This emulates immutability |
| 14 | |
| 15 | |
| 16 | @dataclass(frozen=True) |
| 17 | class BaseConfig: |
| 18 | cloudflared_binary: str |
| 19 | no_autoupdate: bool = True |
| 20 | metrics: str = f'localhost:{METRICS_PORT}' |
| 21 | |
| 22 | def merge_config(self, additional): |
| 23 | config = copy.copy(additional) |
| 24 | config['no-autoupdate'] = self.no_autoupdate |
| 25 | config['metrics'] = self.metrics |
| 26 | return config |
| 27 | |
| 28 | |
| 29 | @dataclass(frozen=True) |
| 30 | class NamedTunnelBaseConfig(BaseConfig): |
| 31 | # The attributes of the parent class are ordered before attributes in this class, |
| 32 | # so we have to use default values here and check if they are set in __post_init__ |
| 33 | tunnel: str = None |
| 34 | credentials_file: str = None |
| 35 | ingress: list = None |
| 36 | |
| 37 | def __post_init__(self): |
| 38 | if self.tunnel is None: |
| 39 | raise TypeError("Field tunnel is not set") |
| 40 | if self.credentials_file is None: |
| 41 | raise TypeError("Field credentials_file is not set") |
| 42 | if self.ingress is None: |
| 43 | raise TypeError("Field ingress is not set") |
| 44 | |
| 45 | def merge_config(self, additional): |
| 46 | config = super(NamedTunnelBaseConfig, self).merge_config(additional) |
| 47 | config['tunnel'] = self.tunnel |
| 48 | config['credentials-file'] = self.credentials_file |
| 49 | # In some cases we want to override default ingress, such as in config tests |
| 50 | if 'ingress' not in config: |
| 51 | config['ingress'] = self.ingress |
| 52 | return config |
| 53 | |
| 54 | |
| 55 | @dataclass(frozen=True) |
| 56 | class NamedTunnelConfig(NamedTunnelBaseConfig): |
| 57 | full_config: dict = None |
| 58 | additional_config: InitVar[dict] = {} |
| 59 | |
| 60 | def __post_init__(self, additional_config): |
| 61 | # Cannot call set self.full_config because the class is frozen, instead, we can use __setattr__ |
| 62 | # https://docs.python.org/3/library/dataclasses.html#frozen-instances |
| 63 | object.__setattr__(self, 'full_config', |
| 64 | self.merge_config(additional_config)) |
| 65 | |
| 66 | def get_url(self): |
| 67 | return "https://" + self.ingress[0]['hostname'] |
| 68 | |
| 69 | |
| 70 | @dataclass(frozen=True) |
| 71 | class ClassicTunnelBaseConfig(BaseConfig): |
| 72 | hostname: str = None |
| 73 | origincert: str = None |
| 74 | |
| 75 | def __post_init__(self): |
| 76 | if self.hostname is None: |
| 77 | raise TypeError("Field tunnel is not set") |
| 78 | if self.origincert is None: |
| 79 | raise TypeError("Field credentials_file is not set") |
| 80 | |
| 81 | def merge_config(self, additional): |
| 82 | config = super(ClassicTunnelBaseConfig, self).merge_config(additional) |
| 83 | config['hostname'] = self.hostname |
| 84 | config['origincert'] = self.origincert |
| 85 | return config |
| 86 | |
| 87 | |
| 88 | @dataclass(frozen=True) |
| 89 | class ClassicTunnelConfig(ClassicTunnelBaseConfig): |
| 90 | full_config: dict = None |
| 91 | additional_config: InitVar[dict] = {} |
| 92 | |
| 93 | def __post_init__(self, additional_config): |
| 94 | # Cannot call set self.full_config because the class is frozen, instead, we can use __setattr__ |
| 95 | # https://docs.python.org/3/library/dataclasses.html#frozen-instances |
| 96 | object.__setattr__(self, 'full_config', |
| 97 | self.merge_config(additional_config)) |
| 98 | |
| 99 | def get_url(self): |
| 100 | return "https://" + self.hostname |
| 101 | |
| 102 | |
| 103 | def build_config_from_env(): |
| 104 | config_path = get_env("COMPONENT_TESTS_CONFIG") |
| 105 | config_content = base64.b64decode( |
| 106 | get_env("COMPONENT_TESTS_CONFIG_CONTENT")).decode('utf-8') |
| 107 | config_yaml = yaml.safe_load(config_content) |
| 108 | |
| 109 | credentials_file = get_env("COMPONENT_TESTS_CREDENTIALS_FILE") |
| 110 | write_file(credentials_file, config_yaml["credentials_file"]) |
| 111 | |
| 112 | origincert = get_env("COMPONENT_TESTS_ORIGINCERT") |
| 113 | write_file(origincert, config_yaml["origincert"]) |
| 114 | |
| 115 | write_file(config_content, config_path) |
| 116 | |
| 117 | |
| 118 | def write_file(content, path): |
| 119 | with open(path, 'w') as outfile: |
| 120 | outfile.write(content) |
| 121 | outfile.close |
| 122 | |
| 123 | |
| 124 | def get_env(env_name): |
| 125 | val = os.getenv(env_name) |
| 126 | if val is None: |
| 127 | raise Exception(f"{env_name} is not set") |
| 128 | return val |
| 129 | |
| 130 | |
| 131 | if __name__ == '__main__': |
| 132 | build_config_from_env() |
| 133 | |